Spaces:
Running on Zero
Running on Zero
| #!/usr/bin/env python3 | |
| """ | |
| Comparify Approach Test Script | |
| ============================= | |
| Tests mobile API endpoints with proper authentication for quick-commerce platforms. | |
| Key insights from research: | |
| 1. Blinkit: Reverse-engineered mobile API (no login needed for search) | |
| 2. Swiggy Instamart: Official MCP server (requires OAuth) | |
| 3. Zepto: BFF gateway with session tokens | |
| This script is separate from the main app - for testing only. | |
| """ | |
| import asyncio | |
| import json | |
| import time | |
| import uuid | |
| from typing import Optional, Dict, Any, List | |
| from dataclasses import dataclass | |
| from curl_cffi.requests import AsyncSession | |
| class Product: | |
| """Standardized product data""" | |
| platform: str | |
| product_id: str | |
| name: str | |
| brand: str | |
| price: float | |
| mrp: float | |
| quantity: str | |
| image_url: str | |
| in_stock: bool | |
| rating: Optional[float] = None | |
| class BlinkitMobileAPI: | |
| """ | |
| Blinkit Mobile API - Reverse-engineered from blinkit-mcp project | |
| Source: https://github.com/yniks/blinkit-mcp | |
| Key findings: | |
| - Uses Cloudflare TLS fingerprinting (need curl_cffi) | |
| - auth_key is a fixed constant (not random UUID) | |
| - No login needed for search/browsing | |
| """ | |
| BASE_URL = "https://blinkit.com" | |
| REQ_KEY = "c0e6868e-1180-400c-be51-f473479f1f0a" # Fixed constant from blinkit-mcp | |
| def __init__(self): | |
| self.session: Optional[AsyncSession] = None | |
| self.auth_key: Optional[str] = None | |
| self.device_id = str(uuid.uuid4())[:16] | |
| self.session_uuid = str(uuid.uuid4()) | |
| # Default to Bangalore coordinates | |
| self.lat = "12.9716" | |
| self.lon = "77.5946" | |
| async def _get_session(self) -> AsyncSession: | |
| """Get or create a curl_cffi session with Chrome impersonation""" | |
| if not self.session: | |
| self.session = AsyncSession(impersonate="chrome150") | |
| return self.session | |
| async def _get_headers(self, content_type: str = "application/json") -> Dict[str, str]: | |
| """Get standard headers for Blinkit API""" | |
| headers = { | |
| "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36", | |
| "Accept": "application/json", | |
| "Accept-Language": "en-US,en;q=0.9", | |
| "app_client": "consumer_web", | |
| "platform": "desktop_web", | |
| "device_id": self.device_id, | |
| "session_uuid": self.session_uuid, | |
| "lat": self.lat, | |
| "lon": self.lon, | |
| "app_version": "52434332", | |
| "web_app_version": "1008010016", | |
| "rn_bundle_version": "1009003012", | |
| } | |
| if self.auth_key: | |
| headers["auth_key"] = self.auth_key | |
| if content_type: | |
| headers["Content-Type"] = content_type | |
| return headers | |
| async def bootstrap(self) -> bool: | |
| """ | |
| Bootstrap: Get auth_key (no login needed) | |
| GET /v2/accounts/auth_key/ with req_key header | |
| """ | |
| print("[Blinkit] Bootstrapping auth_key...") | |
| session = await self._get_session() | |
| headers = await self._get_headers() | |
| headers["req_key"] = self.REQ_KEY | |
| try: | |
| resp = await session.get( | |
| f"{self.BASE_URL}/v2/accounts/auth_key/", | |
| headers=headers, | |
| timeout=10 | |
| ) | |
| if resp.status_code == 200: | |
| data = resp.json() | |
| if data.get("success"): | |
| self.auth_key = data.get("auth_key") | |
| print(f"[Blinkit] ✅ Got auth_key: {self.auth_key[:20]}...") | |
| return True | |
| else: | |
| print(f"[Blinkit] ❌ Bootstrap failed: {data}") | |
| return False | |
| else: | |
| print(f"[Blinkit] ❌ HTTP {resp.status_code}: {resp.text[:200]}") | |
| return False | |
| except Exception as e: | |
| print(f"[Blinkit] ❌ Error: {e}") | |
| return False | |
| async def search(self, query: str, limit: int = 20) -> List[Product]: | |
| """ | |
| Search products | |
| POST /v1/layout/search?q={query}&search_type=type_to_search | |
| """ | |
| print(f"[Blinkit] Searching for: {query}") | |
| if not self.auth_key: | |
| success = await self.bootstrap() | |
| if not success: | |
| return [] | |
| session = await self._get_session() | |
| headers = await self._get_headers() | |
| # Request body (empty for basic search) | |
| body = { | |
| "applied_filters": {}, | |
| "sort": {}, | |
| "postback_meta": {}, | |
| "previous_search_query": "", | |
| "vertical_cards_processed": False | |
| } | |
| try: | |
| resp = await session.post( | |
| f"{self.BASE_URL}/v1/layout/search?q={query}&search_type=type_to_search", | |
| headers=headers, | |
| json=body, | |
| timeout=15 | |
| ) | |
| if resp.status_code == 200: | |
| data = resp.json() | |
| return self._parse_search_results(data, limit) | |
| else: | |
| print(f"[Blinkit] ❌ Search failed: HTTP {resp.status_code}") | |
| print(f"[Blinkit] Response: {resp.text[:500]}") | |
| return [] | |
| except Exception as e: | |
| print(f"[Blinkit] ❌ Search error: {e}") | |
| return [] | |
| def _parse_search_results(self, data: Dict, limit: int) -> List[Product]: | |
| """Parse Blinkit search response into Product objects""" | |
| products = [] | |
| # Blinkit uses server-driven UI with snippets | |
| response = data.get("response", {}) | |
| snippets = response.get("snippets", []) | |
| for snippet in snippets: | |
| if len(products) >= limit: | |
| break | |
| # Look for product data in snippet | |
| snippet_data = snippet.get("data", {}) | |
| # Skip non-product snippets (headers, banners, etc.) | |
| if "product_id" not in snippet_data and "atc_action" not in snippet_data: | |
| continue | |
| # Extract from atc_action (most reliable path) | |
| atc = snippet_data.get("atc_action", {}) | |
| cart_item = atc.get("add_to_cart", {}).get("cart_item", {}) | |
| if cart_item: | |
| try: | |
| # Price is already in INR (integer) | |
| price = cart_item.get("price", 0) | |
| mrp = cart_item.get("mrp", 0) | |
| products.append(Product( | |
| platform="blinkit", | |
| product_id=str(cart_item.get("product_id", "")), | |
| name=cart_item.get("product_name") or snippet_data.get("name", ""), | |
| brand=snippet_data.get("brand") or cart_item.get("brand", ""), | |
| price=float(price) if isinstance(price, (int, float)) else 0, | |
| mrp=float(mrp) if isinstance(mrp, (int, float)) else 0, | |
| quantity=cart_item.get("unit") or snippet_data.get("variant", ""), | |
| image_url=snippet_data.get("image") or snippet_data.get("image_url", ""), | |
| in_stock=cart_item.get("inventory", 0) > 0, | |
| rating=snippet_data.get("rating") | |
| )) | |
| except (ValueError, TypeError) as e: | |
| print(f"[Blinkit] ⚠️ Skipping product: {e}") | |
| return products | |
| class SwiggyInstamartAPI: | |
| """ | |
| Swiggy Instamart API - Using official MCP server | |
| Key findings: | |
| - Official MCP server at mcp.swiggy.com/im | |
| - Requires OAuth authentication | |
| - Has search_products tool available | |
| """ | |
| MCP_URL = "https://mcp.swiggy.com/im" | |
| def __init__(self): | |
| self.session: Optional[AsyncSession] = None | |
| self.access_token: Optional[str] = None | |
| async def _get_session(self) -> AsyncSession: | |
| """Get or create a curl_cffi session""" | |
| if not self.session: | |
| self.session = AsyncSession(impersonate="chrome150") | |
| return self.session | |
| async def search(self, query: str, lat: float = 12.9716, lon: float = 77.5946) -> List[Product]: | |
| """ | |
| Search products using MCP server | |
| Note: This requires OAuth authentication which we don't have. | |
| For testing, we'll try the endpoint and see what happens. | |
| """ | |
| print(f"[Instamart] Searching for: {query}") | |
| session = await self._get_session() | |
| # MCP protocol: initialize first | |
| init_payload = { | |
| "jsonrpc": "2.0", | |
| "id": 1, | |
| "method": "initialize", | |
| "params": { | |
| "protocolVersion": "2024-11-05", | |
| "capabilities": {}, | |
| "clientInfo": { | |
| "name": "test-client", | |
| "version": "1.0.0" | |
| } | |
| } | |
| } | |
| headers = { | |
| "Content-Type": "application/json", | |
| "Accept": "application/json" | |
| } | |
| if self.access_token: | |
| headers["Authorization"] = f"Bearer {self.access_token}" | |
| try: | |
| # First, try to initialize | |
| resp = await session.post( | |
| self.MCP_URL, | |
| headers=headers, | |
| json=init_payload, | |
| timeout=10 | |
| ) | |
| print(f"[Instamart] MCP init response: {resp.status_code}") | |
| if resp.status_code == 200: | |
| # Try to call search_products tool | |
| search_payload = { | |
| "jsonrpc": "2.0", | |
| "id": 2, | |
| "method": "tools/call", | |
| "params": { | |
| "name": "search_products", | |
| "arguments": { | |
| "query": query, | |
| "lat": lat, | |
| "lon": lon | |
| } | |
| } | |
| } | |
| resp2 = await session.post( | |
| self.MCP_URL, | |
| headers=headers, | |
| json=search_payload, | |
| timeout=15 | |
| ) | |
| if resp2.status_code == 200: | |
| return self._parse_mcp_response(resp2.json()) | |
| else: | |
| print(f"[Instamart] ❌ Search failed: {resp2.status_code}") | |
| return [] | |
| else: | |
| print(f"[Instamart] ❌ MCP init failed: {resp.text[:300]}") | |
| return [] | |
| except Exception as e: | |
| print(f"[Instamart] ❌ Error: {e}") | |
| return [] | |
| def _parse_mcp_response(self, data: Dict) -> List[Product]: | |
| """Parse MCP response into Product objects""" | |
| products = [] | |
| # MCP responses are nested | |
| result = data.get("result", {}) | |
| content = result.get("content", []) | |
| for item in content: | |
| if item.get("type") == "text": | |
| try: | |
| text_data = json.loads(item.get("text", "{}")) | |
| items = text_data.get("items", []) | |
| for item_data in items: | |
| products.append(Product( | |
| platform="instamart", | |
| product_id=str(item_data.get("id", "")), | |
| name=item_data.get("name", ""), | |
| brand=item_data.get("brand", ""), | |
| price=float(item_data.get("price", 0)), | |
| mrp=float(item_data.get("mrp", 0)), | |
| quantity=item_data.get("quantity", ""), | |
| image_url=item_data.get("image", ""), | |
| in_stock=item_data.get("available", True), | |
| rating=item_data.get("rating") | |
| )) | |
| except json.JSONDecodeError: | |
| pass | |
| return products | |
| class ZeptoMobileAPI: | |
| """ | |
| Zepto Mobile API - Using BFF gateway | |
| Key findings: | |
| - BFF gateway at bff-gateway.zepto.com | |
| - Returns 202 with empty body for anonymous requests | |
| - Needs proper session/token for search | |
| """ | |
| BASE_URL = "https://bff-gateway.zepto.com" | |
| def __init__(self): | |
| self.session: Optional[AsyncSession] = None | |
| self.session_token: Optional[str] = None | |
| async def _get_session(self) -> AsyncSession: | |
| """Get or create a curl_cffi session""" | |
| if not self.session: | |
| self.session = AsyncSession(impersonate="chrome150") | |
| return self.session | |
| async def search(self, query: str, lat: float = 12.9716, lon: float = 77.5946) -> List[Product]: | |
| """ | |
| Search products using BFF gateway | |
| Note: Zepto's BFF returns 202 with empty body for anonymous requests. | |
| We need to get a session first. | |
| """ | |
| print(f"[Zepto] Searching for: {query}") | |
| session = await self._get_session() | |
| # Try to get a session first | |
| if not self.session_token: | |
| success = await self._get_session_token(lat, lon) | |
| if not success: | |
| return [] | |
| # Try search endpoint | |
| headers = { | |
| "Content-Type": "application/json", | |
| "Accept": "application/json", | |
| "User-Agent": "Zepto/26.8.6 (Android 14; Pixel 8)", | |
| "x-session-token": self.session_token or "" | |
| } | |
| try: | |
| # Try different search endpoints | |
| endpoints = [ | |
| f"{self.BASE_URL}/user-search-service/api/v3/search?q={query}", | |
| f"{self.BASE_URL}/user-search-service/api/v2/search?q={query}", | |
| f"{self.BASE_URL}/inventory-service/api/v1/search?q={query}", | |
| ] | |
| for endpoint in endpoints: | |
| resp = await session.get(endpoint, headers=headers, timeout=10) | |
| if resp.status_code == 200 and resp.text: | |
| try: | |
| data = resp.json() | |
| if data.get("products"): | |
| return self._parse_zepto_response(data) | |
| except json.JSONDecodeError: | |
| pass | |
| print("[Zepto] ❌ All search endpoints returned empty/202") | |
| return [] | |
| except Exception as e: | |
| print(f"[Zepto] ❌ Search error: {e}") | |
| return [] | |
| async def _get_session_token(self, lat: float, lon: float) -> bool: | |
| """Try to get a session token""" | |
| session = await self._get_session() | |
| headers = { | |
| "Content-Type": "application/json", | |
| "Accept": "application/json", | |
| "User-Agent": "Zepto/26.8.6 (Android 14; Pixel 8)", | |
| } | |
| # Try to initialize session | |
| body = { | |
| "lat": lat, | |
| "lon": lon, | |
| "device_id": str(uuid.uuid4())[:16] | |
| } | |
| try: | |
| # Try session initialization endpoint | |
| endpoints = [ | |
| f"{self.BASE_URL}/user-service/api/v1/session", | |
| f"{self.BASE_URL}/location-service/api/v1/resolve", | |
| ] | |
| for endpoint in endpoints: | |
| resp = await session.post(endpoint, headers=headers, json=body, timeout=10) | |
| if resp.status_code == 200: | |
| data = resp.json() | |
| token = data.get("session_token") or data.get("token") | |
| if token: | |
| self.session_token = token | |
| print(f"[Zepto] ✅ Got session token") | |
| return True | |
| print("[Zepto] ⚠️ Could not get session token, trying without...") | |
| return False | |
| except Exception as e: | |
| print(f"[Zepto] ⚠️ Session init error: {e}") | |
| return False | |
| def _parse_zepto_response(self, data: Dict) -> List[Product]: | |
| """Parse Zepto response into Product objects""" | |
| products = [] | |
| items = data.get("products", []) | |
| for item in items: | |
| products.append(Product( | |
| platform="zepto", | |
| product_id=str(item.get("product_id", "")), | |
| name=item.get("name", ""), | |
| brand=item.get("brand", ""), | |
| price=float(item.get("discounted_price", 0)) / 100, # Price in paise | |
| mrp=float(item.get("mrp", 0)) / 100, | |
| quantity=item.get("quantity", ""), | |
| image_url=item.get("image_url", ""), | |
| in_stock=item.get("in_stock", True), | |
| rating=item.get("rating_average") | |
| )) | |
| return products | |
| async def test_all_platforms(query: str = "amul milk"): | |
| """Test all platforms with a search query""" | |
| print("=" * 60) | |
| print(f"Comparify Approach Test - Query: '{query}'") | |
| print("=" * 60) | |
| results = {} | |
| # Test Blinkit | |
| print("\n[1/3] Testing Blinkit Mobile API...") | |
| blinkit = BlinkitMobileAPI() | |
| blinkit_products = await blinkit.search(query) | |
| results["blinkit"] = blinkit_products | |
| print(f"[Blinkit] Found {len(blinkit_products)} products") | |
| # Test Swiggy Instamart | |
| print("\n[2/3] Testing Swiggy Instamart MCP...") | |
| instamart = SwiggyInstamartAPI() | |
| instamart_products = await instamart.search(query) | |
| results["instamart"] = instamart_products | |
| print(f"[Instamart] Found {len(instamart_products)} products") | |
| # Test Zepto | |
| print("\n[3/3] Testing Zepto BFF Gateway...") | |
| zepto = ZeptoMobileAPI() | |
| zepto_products = await zepto.search(query) | |
| results["zepto"] = zepto_products | |
| print(f"[Zepto] Found {len(zepto_products)} products") | |
| # Summary | |
| print("\n" + "=" * 60) | |
| print("RESULTS SUMMARY") | |
| print("=" * 60) | |
| for platform, products in results.items(): | |
| print(f"\n{platform.upper()}: {len(products)} products") | |
| for p in products[:3]: # Show first 3 | |
| print(f" - {p.name[:40]} | ₹{p.price} (MRP: ₹{p.mrp})") | |
| total = sum(len(p) for p in results.values()) | |
| print(f"\n{'=' * 60}") | |
| print(f"TOTAL: {total} products across {sum(1 for p in results.values() if p)} platforms") | |
| print("=" * 60) | |
| return results | |
| if __name__ == "__main__": | |
| # Run the test | |
| asyncio.run(test_all_platforms("amul milk")) | |