from typing import Dict, List, Any, Optional from langchain.tools import tool, StructuredTool try: from pydantic import BaseModel, Field except ImportError: from pydantic.v1 import BaseModel, Field import serpapi import requests from config import settings from database import db, Platform import json @tool("shopping_lite_search", return_direct=False) def shopping_lite_search(category: str, budget_inr: int, location_key: str = "default") -> str: """Search for products using Google Shopping Light API filtered by budget and category. Args: category: Product category (any product type like mobile, laptop, cycle, books, etc.) budget_inr: Maximum budget in Indian Rupees location_key: Location identifier for filtering Returns: JSON string with product candidates including links """ try: import serpapi query = f"{category} under {budget_inr} INR" client = serpapi.Client(api_key=settings.serpapi_api_key) results = client.search({ "engine": "google_shopping_light", "q": query, "gl": "in", "hl": "en", "max_price": budget_inr }) candidates = [] shopping_results = results.get("shopping_results", []) for item in shopping_results: try: # Extract price - google_shopping_light provides extracted_price extracted_price = item.get("extracted_price", 0) if isinstance(extracted_price, (int, float)): price = float(extracted_price) elif isinstance(extracted_price, str): price = float(extracted_price.replace(",", "")) else: continue if price > 0 and price <= budget_inr: candidates.append({ "title": item.get("title", "Unknown Product"), "price": price, "rating": item.get("rating", 0), "reviews": item.get("reviews", 0), "source": item.get("source", "Online Store"), "link": item.get("product_link", ""), # product_link in google_shopping_light "thumbnail": item.get("thumbnail", "") }) except Exception as e: continue if not candidates: return json.dumps({ "success": False, "message": f"No products found under ₹{budget_inr} for {category}", "candidates": [] }) candidates.sort(key=lambda x: (x.get("rating", 0), -x["price"]), reverse=True) return json.dumps({ "success": True, "candidates": candidates[:20], "count": len(candidates) }) except Exception as e: return json.dumps({ "success": False, "error": str(e), "candidates": [] }) @tool("web_search", return_direct=False) def web_search(query: str) -> str: """Perform a web search for additional product context. Args: query: Search query (e.g., "Nokia feature phone under 2000 review") Returns: JSON string with search results summary """ try: params = { "engine": "google", "q": query, "gl": "in", "hl": "en", "num": 5 } client = serpapi.Client(api_key=settings.serpapi_api_key) results = client.search(params) snippets = [] for result in results.get("organic_results", [])[:3]: snippets.append({ "title": result.get("title", ""), "snippet": result.get("snippet", ""), "link": result.get("link", "") }) return json.dumps({ "success": True, "snippets": snippets }) except Exception as e: return json.dumps({ "success": False, "error": str(e), "snippets": [] }) class GetUserProfileInput(BaseModel): """Input schema for get_user_profile tool.""" platform: str = Field(description="Platform name: whatsapp, telegram, or ui") platform_user_id: str = Field(description="User ID on the platform") def _get_user_profile_impl(platform: str, platform_user_id: str) -> str: """Retrieve user profile from database.""" try: user = db.get_user(Platform(platform), platform_user_id) if not user: return json.dumps({ "success": False, "message": "User not found - need to collect profile", "profile": { "id": None, "name": None, "age": None, "location": None, "language": "en" } }) return json.dumps({ "success": True, "profile": { "id": user["id"], "name": user.get("name"), "age": user.get("age"), "location": user.get("location_str"), "language": user.get("language_pref", "en") } }) except Exception as e: return json.dumps({ "success": False, "error": str(e), "profile": None }) get_user_profile = StructuredTool.from_function( func=_get_user_profile_impl, name="get_user_profile", description="Retrieve user profile from database. Check if user exists and what profile fields are missing.", args_schema=GetUserProfileInput, return_direct=False ) # Pydantic models for structured tool inputs class UpdateUserProfileInput(BaseModel): """Input schema for update_user_profile tool.""" platform: str = Field(description="Platform name: whatsapp, telegram, or ui") platform_user_id: str = Field(description="User ID on the platform") name: Optional[str] = Field(default=None, description="User's name (optional)") age: Optional[int] = Field(default=None, description="User's age (optional)") location: Optional[str] = Field(default=None, description="User's location (optional)") def _update_user_profile_impl( platform: str, platform_user_id: str, name: Optional[str] = None, age: Optional[int] = None, location: Optional[str] = None ) -> str: """Update user profile in database.""" try: fields = {} if name: fields['name'] = name if age: fields['age'] = age if location: fields['location_str'] = location user_id = db.upsert_user(Platform(platform), platform_user_id, **fields) return json.dumps({ "success": True, "message": "Profile updated successfully", "user_id": user_id }) except Exception as e: return json.dumps({ "success": False, "error": str(e) }) # Create structured tool with explicit schema update_user_profile = StructuredTool.from_function( func=_update_user_profile_impl, name="update_user_profile", description="Update user profile in database. Use this after collecting user's name, age, or location.", args_schema=UpdateUserProfileInput, return_direct=False ) @tool("check_cache", return_direct=False) def check_cache(category: str, budget_inr: int, location: str = None) -> str: """Check if cached product suggestions exist. Args: category: Product category (mobile, cycle, books) budget_inr: Budget in Indian Rupees location: User location (optional) Returns: JSON string with cached results or empty """ try: location_key = db.make_location_key(location) cached = db.cache_get(category, budget_inr, location_key) if not cached: return json.dumps({ "success": False, "message": "No cache found", "cached": None }) return json.dumps({ "success": True, "cached": { "response_text": cached["response_text"], "picks": cached.get("picks_json", []) } }) except Exception as e: return json.dumps({ "success": False, "error": str(e), "cached": None }) ALL_TOOLS = [ shopping_lite_search, web_search, get_user_profile, update_user_profile, check_cache ]