mcp_client / services /restaurant_service.py
SrikanthNagelli's picture
initial commit
100c46f
"""
Restaurant search service using Foursquare API with intelligent dish recommendations.
"""
import requests
from typing import Optional, List, Dict
from utils.api_config import api_config, CITY_COORDINATES
class DishRecommendationEngine:
"""Intelligent dish recommendation engine based on cuisine types."""
@staticmethod
def get_popular_dishes_by_cuisine(cuisine_type: str, restaurant_name: str) -> List[str]:
"""Generate popular dishes based on cuisine type and restaurant name."""
cuisine_lower = cuisine_type.lower()
name_lower = restaurant_name.lower()
# Coffee shops and cafes
if any(word in cuisine_lower for word in ['coffee', 'cafe', 'bakery']):
if 'peet' in name_lower:
return ['Major Dickason\'s Blend Coffee', 'Espresso Drinks', 'Fresh Pastries', 'Cold Brew']
elif 'starbucks' in name_lower:
return ['Pike Place Roast', 'Pumpkin Spice Latte', 'Frappuccino', 'Cake Pops']
return ['Signature Coffee Blends', 'Fresh Pastries', 'Sandwiches', 'Seasonal Drinks']
# Thai cuisine
elif 'thai' in cuisine_lower:
return ['Pad Thai', 'Green Curry', 'Tom Yum Soup', 'Mango Sticky Rice', 'Massaman Curry']
# Italian cuisine
elif any(word in cuisine_lower for word in ['italian', 'pizza']):
return ['Margherita Pizza', 'Pasta Carbonara', 'Tiramisu', 'Caesar Salad', 'Risotto']
# Chinese cuisine
elif 'chinese' in cuisine_lower:
return ['General Tso\'s Chicken', 'Fried Rice', 'Dumplings', 'Sweet & Sour Pork', 'Lo Mein']
# Japanese cuisine
elif 'japanese' in cuisine_lower or 'sushi' in cuisine_lower:
return ['California Roll', 'Salmon Sashimi', 'Chicken Teriyaki', 'Miso Soup', 'Tempura']
# Mexican cuisine
elif 'mexican' in cuisine_lower:
return ['Fish Tacos', 'Guacamole', 'Quesadillas', 'Burrito Bowl', 'Churros']
# Indian cuisine
elif 'indian' in cuisine_lower:
return ['Butter Chicken', 'Biryani', 'Naan Bread', 'Tikka Masala', 'Samosas']
# French cuisine
elif 'french' in cuisine_lower:
return ['Croissants', 'French Onion Soup', 'Coq au Vin', 'Crème Brûlée', 'Escargot']
# American/General
elif any(word in cuisine_lower for word in ['american', 'burger', 'grill', 'bbq', 'steakhouse']):
return ['Classic Burger', 'BBQ Ribs', 'Mac & Cheese', 'Grilled Salmon', 'Apple Pie']
# Bubble tea shops
elif any(word in cuisine_lower for word in ['bubble tea', 'boba', 'tea']):
return ['Taro Bubble Tea', 'Brown Sugar Milk Tea', 'Fruit Teas', 'Popcorn Chicken', 'Tapioca Pearls']
# Seafood
elif 'seafood' in cuisine_lower:
return ['Fish & Chips', 'Clam Chowder', 'Grilled Salmon', 'Crab Cakes', 'Shrimp Scampi']
# Mediterranean
elif 'mediterranean' in cuisine_lower or 'greek' in cuisine_lower:
return ['Hummus', 'Gyros', 'Greek Salad', 'Baklava', 'Falafel']
# Vietnamese
elif 'vietnamese' in cuisine_lower:
return ['Pho', 'Banh Mi', 'Spring Rolls', 'Vietnamese Coffee', 'Vermicelli Bowl']
# Korean
elif 'korean' in cuisine_lower:
return ['Kimchi', 'Bulgogi', 'Bibimbap', 'Korean BBQ', 'Hotpot']
# Default recommendations
else:
return ['Chef\'s Special', 'House Signature Dish', 'Seasonal Menu Items', 'Daily Specials']
class RestaurantService:
"""Service for searching restaurants using Foursquare API."""
def __init__(self):
self.dish_engine = DishRecommendationEngine()
self.api_key = api_config.foursquare_api_key
def search_restaurants(self, location: str, distance: Optional[float] = None) -> Dict:
"""Search for restaurants in a given location."""
if not self.api_key:
return {"error": "Foursquare API key not configured. Please check server configuration."}
# Clean location input
location = location.lower().replace('restaurants in ', '').strip()
if distance is None:
distance = 1.0
# Convert miles to meters for API
radius = int(distance * 1609.34)
url = "https://api.foursquare.com/v3/places/search"
headers = {
"accept": "application/json",
"Authorization": self.api_key
}
# Check if location matches any known city coordinates
ll_param = None
for city, coords in CITY_COORDINATES.items():
if city in location:
ll_param = coords
break
# Primary search strategy
if ll_param:
params = {
"query": "restaurant",
"ll": ll_param,
"radius": radius,
"categories": "13065", # Restaurant category
"sort": "RATING", # Sort by rating for better results
"limit": 5
}
else:
# Fallback: combined query approach
params = {
"query": f"restaurant {location}",
"radius": radius,
"categories": "13065",
"sort": "RATING",
"limit": 5
}
try:
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
restaurants = response.json().get('results', [])
# If no results with first approach, try alternative
if not restaurants and not ll_param:
location_parts = location.split(',')
if len(location_parts) > 1:
params = {
"query": "restaurant",
"near": location_parts[0].strip(),
"radius": radius,
"categories": "13065",
"sort": "RATING",
"limit": 5
}
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
restaurants = response.json().get('results', [])
if not restaurants:
return {"error": f"No restaurants found in {location}. Try a different location or increase the search radius."}
return self._process_restaurant_results(location, distance, restaurants)
except Exception as e:
return {"error": f"Failed to search restaurants: {str(e)}"}
def _process_restaurant_results(self, location: str, distance: float, restaurants: List[Dict]) -> Dict:
"""Process and format restaurant results."""
result = {
"location": location,
"distance": f"{distance} miles",
"total_found": len(restaurants),
"top_restaurants": []
}
for restaurant in restaurants[:3]: # Limit to top 3
location_data = restaurant.get('location', {})
# Build address more carefully
address_parts = []
if location_data.get('address'):
address_parts.append(location_data['address'])
if location_data.get('locality'):
address_parts.append(location_data['locality'])
if location_data.get('region'):
address_parts.append(location_data['region'])
formatted_address = ", ".join(address_parts) if address_parts else 'Address not available'
# Get cuisine type
categories = restaurant.get('categories', [])
cuisine_type = categories[0].get('name', 'Restaurant') if categories else 'Restaurant'
# Get restaurant name
restaurant_name = restaurant.get('name', 'Unnamed Restaurant')
# Generate popular dishes based on cuisine
popular_dishes = self.dish_engine.get_popular_dishes_by_cuisine(cuisine_type, restaurant_name)
# Format rating
rating = restaurant.get('rating')
rating_display = f"{rating}/10" if rating else 'Not rated'
# Get price level
price_level = restaurant.get('price', 0)
price_display = "$" * max(1, price_level) if price_level else "$"
restaurant_info = {
"name": restaurant_name,
"cuisine_type": cuisine_type,
"address": formatted_address,
"distance": f"{restaurant.get('distance', 0)}m from center",
"rating": rating_display,
"price": price_display,
"description": restaurant.get('description', f"Popular {cuisine_type.lower()} spot with great reviews"),
"recommended_dishes": popular_dishes
}
result['top_restaurants'].append(restaurant_info)
return result
# Global service instance
restaurant_service = RestaurantService()