gift_finder_api.py

"""
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)}")