Spaces:
Sleeping
Sleeping
| 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 | |
| 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": [] | |
| }) | |
| 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 | |
| ) | |
| 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 | |
| ] | |