""" Modified version of gift_finder_api.py with the token removed for security. """ import os import logging import json from typing import Dict, Any, Optional, List from huggingface_client import HuggingFaceClient from gift_finder_model import GiftFinderModel # Set up logging logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class GiftFinderAPI: """ API wrapper for the Gift Finder model that can be used in any application. This API provides a unified interface to use either: 1. Hugging Face Inference API (remote) 2. Local model directly It automatically falls back to the local model if the Hugging Face API is unavailable. """ def __init__(self, hf_token: Optional[str] = None, model_id: str = "mehdirben/gift-finder", use_hf_api: bool = True): """ Initialize the GiftFinderAPI. Args: hf_token: Hugging Face API token (can be None to use local model only) model_id: Model ID on Hugging Face use_hf_api: Whether to try using the Hugging Face API """ self.use_hf_api = use_hf_api and hf_token is not None self.hf_client = None self.local_model = GiftFinderModel() if self.use_hf_api: logger.debug(f"Initializing with Hugging Face API for model: {model_id}") try: self.hf_client = HuggingFaceClient(token=hf_token, model_id=model_id) # Check if HF API is available if not self.hf_client.health_check(): logger.warning("Hugging Face API is not available, falling back to local model.") self.use_hf_api = False except Exception as e: logger.error(f"Error initializing Hugging Face client: {str(e)}") logger.warning("Falling back to local model.") self.use_hf_api = False else: logger.debug("Initializing with local model only.") def process_conversation(self, conversation: str) -> Dict[str, Any]: """ Process a conversation to extract gift-related information and generate a search query. Args: conversation: The conversation text to analyze Returns: Dict containing extracted_data and gift_search_query """ logger.debug(f"Processing conversation: {conversation[:100]}...") try: # Try using Hugging Face API if enabled if self.use_hf_api and self.hf_client: try: logger.debug("Using Hugging Face API") return self.hf_client.process_conversation(conversation) except Exception as e: logger.error(f"Error with Hugging Face API: {str(e)}") logger.warning("Falling back to local model") # Use local model logger.debug("Using local model") return self.local_model.process_conversation(conversation) except Exception as e: logger.error(f"Error processing conversation: {str(e)}") raise def extract_gift_data(self, conversation: str) -> Dict[str, Any]: """ Extract gift-related data from a conversation. Args: conversation: The conversation text to analyze Returns: Dict containing extracted gift data """ result = self.process_conversation(conversation) return result["extracted_data"] def generate_gift_query(self, conversation: str) -> str: """ Generate a gift search query from a conversation. Args: conversation: The conversation text to analyze Returns: Gift search query string """ result = self.process_conversation(conversation) return result["gift_search_query"] def get_gift_recommendations(self, conversation: str, num_recommendations: int = 3, search_api_key: Optional[str] = None) -> List[Dict[str, Any]]: """ Get gift recommendations based on a conversation. This method can use SerpAPI to search for real gift products if a search_api_key is provided. Otherwise, it falls back to example recommendations. Args: conversation: The conversation text to analyze num_recommendations: Number of recommendations to return search_api_key: API key for SerpAPI (optional) Returns: List of gift recommendations """ # Generate search query from the conversation query = self.generate_gift_query(conversation) extracted_data = self.extract_gift_data(conversation) logger.debug(f"Generated search query: {query}") # Parse budget information for price range filtering price_min = None price_max = None budget = extracted_data.get("budget", "") if "under" in budget.lower() and "$" in budget: try: # Extract the maximum price value price_str = budget.split("$")[1].split()[0].replace(",", "") price_max = float(price_str) except (IndexError, ValueError): pass # If search_api_key is provided, use SerpAPI for real product search if search_api_key: try: logger.debug("Using SerpAPI for real product search") from gift_search_engine import GiftSearchEngine # Initialize the search engine with the provided API key search_engine = GiftSearchEngine(api_key=search_api_key) # Search for gifts recommendations = search_engine.search_gifts( query=query, num_results=num_recommendations, price_min=price_min, price_max=price_max ) if recommendations: logger.debug(f"Found {len(recommendations)} recommendations using SerpAPI") return recommendations else: logger.warning("No recommendations found with SerpAPI, falling back to examples") except Exception as e: logger.error(f"Error using SerpAPI: {str(e)}") logger.warning("Falling back to example recommendations") # Fallback to example recommendations based on extracted data logger.debug("Using example gift recommendations") recommendations = [] if extracted_data["occasion"] == "birthday": recommendations.append({ "name": "Birthday Gift Card", "price": "$25.00", "url": "https://example.com/gift-card", "description": "A versatile gift card for any birthday." }) if "gaming" in extracted_data.get("recipient_interests", ""): recommendations.append({ "name": "Gaming Mouse Pad", "price": "$19.99", "url": "https://example.com/gaming-mouse-pad", "description": "Premium gaming mouse pad with RGB lighting." }) if "basketball" in extracted_data.get("recipient_interests", ""): recommendations.append({ "name": "Mini Basketball Hoop", "price": "$29.99", "url": "https://example.com/mini-basketball-hoop", "description": "Indoor mini basketball hoop for door or wall." }) # Add generic recommendations if we don't have enough generic_recommendations = [ { "name": "Wireless Earbuds", "price": "$49.99", "url": "https://example.com/wireless-earbuds", "description": "High-quality wireless earbuds with noise cancellation." }, { "name": "Portable Charger", "price": "$34.99", "url": "https://example.com/portable-charger", "description": "10000mAh portable charger for smartphones and tablets." }, { "name": "Coffee Mug Set", "price": "$24.99", "url": "https://example.com/coffee-mug-set", "description": "Set of 4 ceramic coffee mugs in assorted colors." } ] # Add generic recommendations until we have enough while len(recommendations) < num_recommendations and generic_recommendations: recommendations.append(generic_recommendations.pop(0)) return recommendations[:num_recommendations] # Example usage if __name__ == "__main__": # Get token from environment token = os.environ.get("HF_TOKEN") # Create API wrapper api = GiftFinderAPI(hf_token=token, use_hf_api=True) # Example conversation conversation = """ USER: I need to find a gift for my nephew's birthday. He's 12 and loves basketball and video games. My budget is around $30. ASSISTANT: I'd be happy to help you find a birthday gift for your nephew who enjoys basketball and video games. USER: He especially likes playing Minecraft and Fortnite. And for basketball, he's a big fan of the Lakers. """ try: # Process the conversation print("Processing conversation...") result = api.process_conversation(conversation) print("\nExtracted Gift Data:") print(json.dumps(result["extracted_data"], indent=2)) print("\nGenerated Gift Search Query:") print(result["gift_search_query"]) # Get recommendations print("\nGift Recommendations:") recommendations = api.get_gift_recommendations(conversation) for i, rec in enumerate(recommendations, 1): print(f"{i}. {rec['name']} - {rec['price']}") print(f" {rec['description']}") print(f" {rec['url']}") print() except Exception as e: print(f"Error using Gift Finder API: {str(e)}")