#!/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 @dataclass 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"))