# app.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ SaveNest — 'Fetch Everything' Crawler API ========================================= - Enumerates product URLs via sitemaps (Shopify / Woo + generic sitemap discovery). - Optional fallback BFS enumerator for stores without sitemaps (lightweight, bounded). - Scrapes product details with JSON-LD first; CSS fallback; optional Playwright (Firefox) for JS-heavy pages (Daraz etc). - Stateless on the server: supports paginated crawling (cursor/limit). You can persist the results in Firebase/Firestore from your app. Endpoints: - GET /health - GET / -> intro JSON - GET /crawl -> enumerate + scrape entire store (paginated) params: store : domain OR key (e.g. "alfatah.pk", "daraz.pk") limit : number of product pages to fetch this call (default 100, max 500) cursor : resume offset (int; default 0) mode : "full" (default) or "urls" (only URLs, no scraping) use_js : 0/1 override for Playwright for this request (default: env USE_PLAYWRIGHT) - GET /search -> per-query live search (kept from previous build) Designed for HuggingFace Spaces free tier: - Restrict limit per call to avoid timeouts; use `cursor` to continue next page. - Uses a gentle rate limiter per host. """ import asyncio, time, re, json, os, gzip, io from typing import Optional, Dict, Any, List, Tuple, Iterable from urllib.parse import urljoin, urlparse, quote_plus from dataclasses import dataclass from fastapi.middleware.cors import CORSMiddleware import os import httpx from bs4 import BeautifulSoup from fastapi import FastAPI, HTTPException, Query from fastapi.responses import JSONResponse, HTMLResponse from rapidfuzz import fuzz import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', datefmt='%H:%M:%S' ) logger = logging.getLogger(__name__) # ---------------------- Runtime config (via env) ---------------------- USE_PLAYWRIGHT_DEFAULT = os.getenv("USE_PLAYWRIGHT", "1") == "1" # enable JS fallback by default now PER_HOST_GAP = float(os.getenv("PER_HOST_GAP", "2")) # politeness per host (seconds) REQUEST_TIMEOUT = float(os.getenv("REQUEST_TIMEOUT", "30")) # seconds TTL_SECONDS = int(os.getenv("TTL_SECONDS", "180")) # page TTL cache MAX_CONCURRENCY = int(os.getenv("MAX_CONCURRENCY", "5")) # parallel HTTP requests CRAWL_LIMIT_CAP = int(os.getenv("CRAWL_LIMIT_CAP", "500")) # per-call max items MAX_FALLBACK_BFS_PAGES = int(os.getenv("MAX_FALLBACK_BFS_PAGES", "50")) # BFS page cap if no sitemap # ---------------------- Store registry (extend as needed) ------------ # Each entry supports: # base: canonical base url # platform: "shopify" | "woo" | "custom" # product_link_patterns: regex patterns to recognize product pages # product_fallback: CSS selectors for title/price if no JSON-LD # sitemap_hints: list of potential sitemap endpoints # bfs_seeds: optional seeds for BFS (when no sitemap is available) # ---------------------- Store registry (extend as needed) ------------ STORES: List[Dict[str, Any]] = [ # Existing — keep as-is {"base": "https://alfatah.pk", "platform": "shopify"}, # NEW STORES from API #2 { "base": "https://qne.com.pk", "platform": "custom", "search_url_template": "{base}/search?type=product&q={query}", "product_link_patterns": [r"/products?/[-a-z0-9]+", r"/p/[-a-z0-9]+"], "product_fallback": { "title": "h1.product-title, h1.entry-title, .product-name", "price": ".price, .amount, span.price, .woocommerce-Price-amount", "image": ".product-image img, img[src*='product'], .attachment-woocommerce_thumbnail" } }, { "base": "https://springs.com.pk", "platform": "custom", "search_url_template": "{base}/search?q={query}", "product_link_patterns": [r"/products?/[-a-z0-9]+"], "product_fallback": { "title": "h1.product_title, h1.entry-title, h1[itemprop='name']", "price": ".price ins .amount, .price > span, .woocommerce-Price-amount", "image": ".woocommerce-product-gallery__image img, img.wp-post-image, img.attachment-product" } }, { "base": "https://vmart.pk", "platform": "custom", "search_url_template": "{base}/search?q={query}", "product_link_patterns": [r"/products?/[-a-z0-9]+"], "product_fallback": { "title": "h1.product-title, .product-name, h1[itemprop='name']", "price": ".price, .amount, span.price-current", "image": "img.product-image, .main-image img, img[src*='product']" } }, { "base": "https://grocerapp.pk", "platform": "custom", "search_url_template": "{base}/search?q={query}", "sitemap_hints": [ "https://grocerapp.pk/sitemap_index.xml", "https://grocerapp.pk/sitemap_categories.xml", "https://grocerapp.pk/sitemap_subcategories.xml", ], "product_link_patterns": [r"/products/[-a-z0-9]+-\d+/?$"], "product_fallback": { "title": "h1.product-title, .product-detail h1, h1[itemprop='name']", "price": ".price-current, .amount, .sale-price", "image": ".product-image img, img.main-product-image, img[src*='product']" } }, ] STORE_DISPLAY_NAMES = { "alfatah.pk": "Al-Fatah", "qne.com.pk": "QnE", "springs.com.pk": "Springs Store", "vmart.pk": "Vmart", "grocerapp.pk": "GrocerApp", } # ---------------------- Helpers -------------------------------------- def extract_og_image(soup: BeautifulSoup, page_url: str) -> Optional[str]: # 1) meta = soup.find("meta", attrs={"property": "og:image"}) or soup.find("meta", attrs={"name": "og:image"}) if meta and meta.get("content"): return urljoin(page_url, meta["content"].strip()) # 2) link = soup.find("link", rel=lambda v: v and "image_src" in v) if link and link.get("href"): return urljoin(page_url, link["href"].strip()) return None def d(u: str) -> str: return urlparse(u).netloc.lower().replace("www.", "") def abs_url(base: str, href: str) -> str: return urljoin(base.rstrip("/") + "/", href) def _norm(s: str) -> str: return re.sub(r"[^a-z0-9\s]+", " ", (s or "").lower()).strip() def clean_price(text: str) -> Optional[float]: if not text: return None t = text.replace(",", "") m = re.search(r"(\d+(?:\.\d{1,2})?)", t) return float(m.group(1)) if m else None def guess_store(domain_or_key: str) -> Optional[Dict[str, Any]]: key = domain_or_key.strip().lower() for s in STORES: if key in d(s["base"]) or key in s["base"].lower(): return s return None # politeness limiter _last_hit: Dict[str, float] = {} async def rate_limit(host: str, min_gap: float = PER_HOST_GAP): last = _last_hit.get(host, 0.0) dt = time.time() - last if dt < min_gap: await asyncio.sleep(min_gap - dt) _last_hit[host] = time.time() BROWSER_HEADERS = { "User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36"), "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8", "Accept-Language": "en-GB,en-US;q=0.9,en;q=0.8", "Cache-Control": "no-cache", "Pragma": "no-cache", } # in-memory TTL cache _cache: Dict[str, Tuple[float, Optional[str]]] = {} def cache_get(url: str) -> Optional[str]: v = _cache.get(url) if not v: return None ts, body = v if time.time() - ts > TTL_SECONDS: _cache.pop(url, None) return None return body def cache_put(url: str, body: Optional[str]): _cache[url] = (time.time(), body) async def fetch_text(client: httpx.AsyncClient, url: str, referer: Optional[str] = None) -> Optional[str]: cached = cache_get(url) if cached is not None: return cached try: await rate_limit(urlparse(url).netloc) headers = dict(BROWSER_HEADERS) if referer: headers["Referer"] = referer r = await client.get(url, headers=headers, timeout=REQUEST_TIMEOUT, follow_redirects=True) if r.status_code != 200: cache_put(url, None) return None # gunzip transparently if needed content = r.content if r.headers.get("Content-Encoding", "").lower() == "gzip" or url.endswith(".gz"): try: content = gzip.decompress(content) except Exception: pass text = content.decode(r.encoding or "utf-8", errors="ignore") cache_put(url, text) return text except Exception: cache_put(url, None) return None # optional Playwright (Firefox) renderer async def fetch_js(url: str) -> Optional[str]: try: from playwright.async_api import async_playwright except Exception: return None try: async with async_playwright() as p: browser = await p.firefox.launch(headless=True) page = await browser.new_page(user_agent=BROWSER_HEADERS["User-Agent"]) await page.goto(url, wait_until="domcontentloaded", timeout=int((REQUEST_TIMEOUT+10) * 1000)) await page.wait_for_timeout(1200) html = await page.content() await browser.close() return html except Exception: return None # ---------------------- Parsing product pages ------------------------- @dataclass class Offer: source: str title: str price: Optional[float] currency: Optional[str] url: str image: Optional[str] = None in_stock: Optional[bool] = None image_url: Optional[str] = None SERVICE_BLACKLIST = [ "service", "services", "installation", "repair", "fixing", "cleaning", "alignment", "movers", "packers", "examination", "inspection", "consultation" ] def looks_like_service(title: str) -> bool: t = (title or "").lower() return any(w in t for w in SERVICE_BLACKLIST) def parse_ldjson(html: str, page_url: str) -> List[Offer]: soup = BeautifulSoup(html or "", "html.parser") out: List[Offer] = [] for tag in soup.find_all("script", type="application/ld+json"): raw = tag.string or "" if not raw.strip(): continue try: data = json.loads(raw) except Exception: continue blocks = data if isinstance(data, list) else [data] for b in blocks: if not isinstance(b, dict): continue typ = b.get("@type") is_product = (isinstance(typ, list) and "Product" in typ) or (typ == "Product") if not is_product: continue name = (b.get("name") or "").strip() url = b.get("url") or page_url # ✅ Improved image extraction img = b.get("image") if isinstance(img, dict): img = img.get("url") or img.get("@id") elif isinstance(img, list) and img: img = img[0] image_url = str(img) if isinstance(img, str) else None # Handle offers offs = b.get("offers") arr = offs if isinstance(offs, list) else [offs] if offs else [] if not arr: # ✅ Fallback: check priceSpecification or aggregateOffer ps = b.get("priceSpecification") if isinstance(ps, dict): price = clean_price(str(ps.get("price") or "")) cur = ps.get("priceCurrency") if name and price is not None: out.append(Offer( source=d(page_url), title=name, price=price, currency=cur, url=url, image_url=image_url )) agg = b.get("aggregateOffer") if isinstance(agg, dict): price = clean_price(str(agg.get("lowPrice") or agg.get("price") or "")) cur = agg.get("priceCurrency") if name and price is not None: out.append(Offer( source=d(page_url), title=name, price=price, currency=cur, url=url, image_url=image_url )) for o in arr: if not isinstance(o, dict): continue price = o.get("price") if price is None and isinstance(o.get("priceSpecification"), dict): price = o["priceSpecification"].get("price") currency = o.get("priceCurrency") or (o.get("priceSpecification") or {}).get("priceCurrency") avail = o.get("availability") in_stock = None if isinstance(avail, str): in_stock = "OutOfStock" not in avail and "SoldOut" not in avail p = clean_price(str(price) if price is not None else "") if name and p is not None: out.append(Offer( source=d(page_url), title=name, price=p, currency=currency, url=url, in_stock=in_stock, image_url=image_url )) return out def css_text(soup: BeautifulSoup, sel: Optional[str]) -> Optional[str]: if not sel: return None node = soup.select_one(sel) return node.get_text(" ", strip=True) if node else None def css_attr(soup: BeautifulSoup, sel: Optional[str], attr: str) -> Optional[str]: if not sel: return None node = soup.select_one(sel) return node.get(attr) if node else None def parse_product_fallback(html: str, page_url: str, fallback: Dict[str, str]) -> Optional[Offer]: soup = BeautifulSoup(html or "", "html.parser") title = css_text(soup, fallback.get("title")) price_text = css_text(soup, fallback.get("price")) price = clean_price(price_text or "") # ✅ Improved image fallback image_url = css_attr(soup, fallback.get("image", "img[src]"), "src") if not image_url: image_url = extract_og_image(soup, page_url) if title and price is not None: return Offer( source=d(page_url), title=title, price=price, currency=None, url=page_url, image_url=image_url # ✅ Updated field ) return None async def parse_product( client: httpx.AsyncClient, store: Dict[str, Any], url: str, use_js: bool, fetch_func=fetch_text # ← NEW: custom fetcher, defaults to global one for backward compatibility ) -> List[Offer]: html = await fetch_func(client, url, referer=store["base"]) if not html and use_js: html = await fetch_js(url) if not html: return [] offers = parse_ldjson(html, url) if offers: return offers # CSS fallback by platform if store["platform"] == "shopify": fallback = { "title": "h1.product__title, h1.product-title, h1.page-title", "price": ".price-item--regular, .price .price__regular, span.price", "image": "img[src]" } elif store["platform"] == "woo": fallback = { "title": "h1.product_title, h1.entry-title", "price": "p.price, span.price, bdi", "image": "img[src]" } else: fallback = store.get("product_fallback", { "title": "h1, h1.product-title", "price": "span.price, .price", "image": "img[src]" }) one = parse_product_fallback(html, url, fallback) if fallback else None return [one] if one else [] # ---------------------- Sitemap discovery & enumeration --------------- SITEMAP_HINTS_DEFAULT = [ "/sitemap.xml", "/sitemap_index.xml", "/sitemap-index.xml", ] async def discover_sitemaps(client: httpx.AsyncClient, base: str, store: Dict[str, Any]) -> List[str]: results: List[str] = [] # robots.txt robots = await fetch_text(client, abs_url(base, "/robots.txt")) if robots: for line in robots.splitlines(): if line.lower().startswith("sitemap:"): sm = line.split(":", 1)[1].strip() if sm and sm not in results: results.append(sm) # hints from store for h in store.get("sitemap_hints", []): results.append(h) # defaults for h in SITEMAP_HINTS_DEFAULT: results.append(abs_url(base, h)) # uniq, keep order seen = set(); out = [] for u in results: if u not in seen: seen.add(u); out.append(u) return out def _extract_xml_urls(xml_text: str) -> List[str]: urls: List[str] = [] soup = BeautifulSoup(xml_text or "", "xml") # needs 'lxml' or bs4's xml parser for loc in soup.find_all("loc"): if loc and loc.text: urls.append(loc.text.strip()) return urls async def iter_sitemap_tree(client: httpx.AsyncClient, sitemap_url: str, max_nodes: int = 1000) -> List[str]: """Return flattened list of URLs from a (possibly nested) sitemap index.""" urls: List[str] = [] seen: set = set() queue: List[str] = [sitemap_url] while queue and len(seen) < max_nodes: u = queue.pop(0) if u in seen: continue seen.add(u) txt = await fetch_text(client, u) if not txt: continue locs = _extract_xml_urls(txt) if any(u2.endswith(".xml") or "/sitemap" in u2 for u2 in locs): # child sitemaps for c in locs: if c not in seen and len(seen) + len(queue) < max_nodes: queue.append(c) else: urls.extend(locs) return urls def _match_product(url: str, store: Dict[str, Any]) -> bool: pats = store.get("product_link_patterns") if pats: return any(re.search(p, url) for p in pats) # platform defaults if store["platform"] == "shopify": return re.search(r"/products?/[-a-z0-9]+/?$", url) is not None if store["platform"] == "woo": return re.search(r"/product/[-a-z0-9]+/?$", url) is not None return True # best effort async def enumerate_products_via_sitemaps(client: httpx.AsyncClient, store: Dict[str, Any]) -> List[str]: base = store["base"] roots = await discover_sitemaps(client, base, store) all_urls: List[str] = [] for sm in roots: items = await iter_sitemap_tree(client, sm, max_nodes=5000) all_urls.extend(items) # filter to same domain + product patterns dd = d(base) prods = [] seen = set() for u in all_urls: if d(u).endswith(dd) and _match_product(u, store): if u not in seen: seen.add(u); prods.append(u) return prods # ---------------------- Fallback BFS (bounded) ------------------------ async def enumerate_products_via_bfs(client: httpx.AsyncClient, store: Dict[str, Any]) -> List[str]: base = store["base"].rstrip("/") seeds = store.get("bfs_seeds") or [base] dd = d(base) to_visit = list({abs_url(base, s) for s in seeds}) visited = set() found: List[str] = [] pages = 0 while to_visit and pages < MAX_FALLBACK_BFS_PAGES and len(found) < 8000: url = to_visit.pop(0) if url in visited: continue visited.add(url) pages += 1 html = await fetch_text(client, url, referer=base) if not html: continue soup = BeautifulSoup(html, "html.parser") # harvest links for a in soup.find_all("a", href=True): href = abs_url(base, a["href"]) if d(href).endswith(dd): if _match_product(href, store): if href not in found: found.append(href) elif href not in visited and len(to_visit) + pages < MAX_FALLBACK_BFS_PAGES * 20: # enqueue further category/listing pages conservatively if re.search(r"(category|catalog|search|page=|/c/|/s/)", href, re.I): to_visit.append(href) return found # ---------------------- Query expansion (reused search) --------------- async def harvest_shopify_suggestions(client: httpx.AsyncClient, stores: List[dict], query: str, limit_per_store: int = 10) -> List[str]: suggestions: List[str] = [] for s in stores: if s.get("platform") != "shopify": continue base = s["base"].rstrip("/") sug_url = f"{base}/search/suggest.json?q={quote_plus(query)}&resources[type]=product&resources[limit]={limit_per_store}" txt = await fetch_text(client, sug_url, referer=base) if not txt: continue try: data = json.loads(txt) prods = data.get("resources", {}).get("results", {}).get("products", []) for p in prods: title = p.get("title") or p.get("handle") or "" if title: suggestions.append(title.strip()) except Exception: continue # de-dupe case-insensitive seen = set(); uniq = [] for s in suggestions: k = s.lower() if k not in seen: seen.add(k); uniq.append(s) return uniq # ---------------------- FastAPI app ----------------------------------- app = FastAPI(title="SaveNest Everything Crawler API", version="2.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], # or restrict to your app’s domain(s) allow_methods=["*"], allow_headers=["*"], ) @app.get("/", response_class=HTMLResponse) async def root(): return """ SaveNest Crawler API
🛒
SaveNest API
v2.0 Live
📖 Docs
🕷️ Grocery Price Intelligence

Crawl. Compare.
Save Smarter.

Real-time product price scraping across Pakistan's top grocery stores. Sitemap enumeration, BFS fallback, JSON-LD parsing — all in one API.

5 Stores
500 Max / Call
3 Endpoints
PKR Currency
Supported Stores
Al-Fatah alfatah.pk
QnE qne.com.pk
Springs springs.com.pk
Vmart vmart.pk
GrocerApp grocerapp.pk
Endpoints
GET /health API status check ▾

Returns API operational status.

// Response { "ok": true }
GET /search Live product search across all stores ▾

Search for a product across all registered stores simultaneously. Returns deduplicated, price-sorted results.

ParameterTypeDescription
qrequired string Product keyword (e.g., "milk", "bread", "laptop")
limitoptional integer Max results (1–100, default: 40)
// GET /search?q=milk&limit=10 { "query": "milk", "count": 4, "offers": [ { "source": "Al-Fatah", "title": "Olpers Full Cream Milk 1L", "price": 285.0, "currency": "PKR", "url": "https://alfatah.pk/products/...", "image": "https://...", "in_stock": true } ] }
GET /crawl Full store enumeration + scraping ▾

Enumerate ALL product URLs from a store via sitemaps (BFS fallback) and optionally scrape details. Use cursor for pagination.

ParameterTypeDescription
storerequired string Domain key, e.g. "alfatah.pk"
limitoptional integer Items per call (1–500, default: 100)
cursoroptional integer Resume offset for pagination (default: 0)
modeoptional string "full" (scrape) or "urls" (enumerate only)
use_jsoptional 0|1 Override Playwright JS rendering
// GET /crawl?store=alfatah.pk&limit=50&cursor=0 { "store": "alfatah.pk", "count": 48, "items": [ /* array of product objects */ ], "next_cursor": 50, "total_urls": 1240, "enumeration": "sitemap_or_bfs", "js_fallback": true }
Live API Tester

Check if the API is running and operational.

""" @app.get("/health") async def health(): return {"ok": True} # SaveNest app.py @app.get("/documentation") async def documentation(): from fastapi.responses import RedirectResponse return RedirectResponse("https://aero-woad.vercel.app/savenest-docs.html") @app.get("/crawl") async def crawl( store: str = Query(..., description="Domain or key, e.g., 'alfatah.pk', 'daraz.pk'"), limit: int = Query(100, ge=1, le=CRAWL_LIMIT_CAP), cursor: int = Query(0, ge=0), mode: str = Query("full", pattern="^(full|urls)$"), use_js: Optional[int] = Query(None, description="Override USE_PLAYWRIGHT (1 or 0)"), ): """ Enumerate ALL product URLs for a store (via sitemaps, else BFS) and optionally scrape details. Pagination: call repeatedly with `cursor=next_cursor` until next_cursor is None. """ s = guess_store(store) if not s: raise HTTPException(400, f"Unknown store: {store}") use_js_effective = USE_PLAYWRIGHT_DEFAULT if use_js is None else (use_js == 1) async with httpx.AsyncClient(follow_redirects=True, timeout=REQUEST_TIMEOUT) as client: # 1) enumerate product URLs urls = await enumerate_products_via_sitemaps(client, s) if not urls: urls = await enumerate_products_via_bfs(client, s) if not urls: return {"store": store, "count": 0, "items": [], "next_cursor": None, "enumeration": "none"} # 2) paginate start = cursor end = min(len(urls), start + limit) batch = urls[start:end] if mode == "urls": return { "store": store, "count": len(batch), "items": [{"url": u} for u in batch], "next_cursor": (end if end < len(urls) else None), "total_urls": len(urls), "enumeration": "sitemap_or_bfs" } # 3) scrape product pages (concurrently, politely) sem = asyncio.Semaphore(MAX_CONCURRENCY) async def _one(u: str): async with sem: try: return await parse_product(client, s, u, use_js_effective) except Exception: return [] tasks = [_one(u) for u in batch] parsed = await asyncio.gather(*tasks, return_exceptions=True) # flatten + clean offers: List[Offer] = [] for r in parsed: if isinstance(r, Exception) or not r: continue offers.extend(r) def _ok(o: Offer) -> bool: return o.title and (o.price is not None) and not looks_like_service(o.title) items = [{ "source": STORE_DISPLAY_NAMES.get(o.source, o.source.replace(".pk", "").title()), "title": o.title, "price": o.price, "currency": o.currency or "PKR", "url": o.url, "image": o.image_url, "in_stock": o.in_stock, } for o in offers if _ok(o)] # de-duplicate by normalized title+source keep cheapest keyd: Dict[Tuple[str, str], Dict[str, Any]] = {} for it in items: k = (it["source"], _norm(it["title"])) if k not in keyd or (it["price"] or 1e18) < (keyd[k]["price"] or 1e18): keyd[k] = it items = list(keyd.values()) items.sort(key=lambda x: (x["price"] if x["price"] is not None else 1e18, x["title"].lower())) return { "store": store, "count": len(items), "items": items, "next_cursor": (end if end < len(urls) else None), "total_urls": len(urls), "enumeration": "sitemap_or_bfs", "js_fallback": use_js_effective, } # (kept) lightweight live search from previous build, handy for ad-hoc queries @app.get("/search") async def search( q: str = Query(..., description="Product keyword (e.g., 'bread', 'milk', 'laptop')"), limit: int = Query(40, ge=1, le=100), ): """ Search for products across all stores. ✅ FIX #1: Per-request local cache (no global pollution) ✅ FIX #2: Proper search_url_template usage ✅ Comprehensive logging for debugging """ request_id = f"{int(time.time() * 1000) % 100000}" logger.info(f"[{request_id}] ========== NEW SEARCH REQUEST ==========") logger.info(f"[{request_id}] Query: '{q}' | Limit: {limit}") q = q.strip() if not q: raise HTTPException(status_code=400, detail="Missing query") # 🔧 FIX #1: Per-request local cache and rate limiter local_cache: Dict[str, Tuple[float, Optional[str]]] = {} local_last_hit: Dict[str, float] = {} logger.info(f"[{request_id}] 🆕 Fresh local cache created (global cache has {len(_cache)} entries but will be ignored for product parsing)") logger.info(f"[{request_id}] 🆕 Created fresh local cache for this request") def cache_get(url: str) -> Optional[str]: v = local_cache.get(url) if not v: return None ts, body = v if time.time() - ts > TTL_SECONDS: local_cache.pop(url, None) logger.debug(f"[{request_id}] ⏰ Cache expired for {url}") return None logger.debug(f"[{request_id}] 💾 Cache hit for {url}") return body def cache_put(url: str, body: Optional[str]): local_cache[url] = (time.time(), body) status = "✅ cached" if body else "❌ cached as None" logger.debug(f"[{request_id}] {status}: {url}") async def rate_limit_local(host: str, min_gap: float = PER_HOST_GAP): last = local_last_hit.get(host, 0.0) dt = time.time() - last if dt < min_gap: sleep_time = min_gap - dt logger.debug(f"[{request_id}] ⏳ Rate limiting {host}: sleeping {sleep_time:.2f}s") await asyncio.sleep(sleep_time) local_last_hit[host] = time.time() async def fetch_text_local(client: httpx.AsyncClient, url: str, referer: Optional[str] = None) -> Optional[str]: cached = cache_get(url) if cached is not None: return cached try: await rate_limit_local(urlparse(url).netloc) headers = dict(BROWSER_HEADERS) if referer: headers["Referer"] = referer logger.debug(f"[{request_id}] 🌐 Fetching: {url}") r = await client.get(url, headers=headers, timeout=REQUEST_TIMEOUT, follow_redirects=True) if r.status_code != 200: logger.warning(f"[{request_id}] ⚠️ HTTP {r.status_code} for {url}") cache_put(url, None) return None content = r.content if r.headers.get("Content-Encoding", "").lower() == "gzip" or url.endswith(".gz"): try: content = gzip.decompress(content) except Exception as e: logger.warning(f"[{request_id}] ⚠️ Gunzip failed for {url}: {e}") text = content.decode(r.encoding or "utf-8", errors="ignore") logger.debug(f"[{request_id}] ✅ Fetched {len(text)} bytes from {url}") cache_put(url, text) return text except Exception as e: logger.error(f"[{request_id}] ❌ Fetch error for {url}: {e}") cache_put(url, None) return None async with httpx.AsyncClient(follow_redirects=True, timeout=REQUEST_TIMEOUT) as client: async def _links_for_store(s: Dict[str, Any]) -> List[str]: store_name = d(s["base"]) logger.info(f"[{request_id}] 🏪 Processing store: {store_name} ({s['platform']})") links: List[str] = [] base = s["base"].rstrip("/") # ---------- Shopify ---------- if s["platform"] == "shopify": # 1) Try suggest.json sug_url = f"{base}/search/suggest.json?q={quote_plus(q)}&resources[type]=product&resources[limit]=10" logger.info(f"[{request_id}] → Shopify suggest: {sug_url}") txt = await fetch_text_local(client, sug_url, referer=base) if txt: logger.info(f"[{request_id}] ✅ Suggest API response: {len(txt)} bytes") try: data = json.loads(txt) prods = data.get("resources", {}).get("results", {}).get("products", []) logger.info(f"[{request_id}] 📦 Found {len(prods)} products in suggest JSON") for p in prods: if "handle" in p: product_url = abs_url(base, f"/products/{p['handle']}") links.append(product_url) except Exception as e: logger.error(f"[{request_id}] ❌ JSON parse error: {e}") else: logger.warning(f"[{request_id}] ⚠️ No response from suggest API") # 2) Fallback: ?view=json if not links: json_url = f"{base}/search?q={quote_plus(q)}&view=json" logger.info(f"[{request_id}] → Shopify JSON view fallback: {json_url}") txt = await fetch_text_local(client, json_url, referer=base) if txt and txt.strip().startswith("{"): try: data = json.loads(txt) prods = data.get("products", []) logger.info(f"[{request_id}] 📦 Found {len(prods)} products in JSON view") for prod in prods: if "url" in prod: links.append(abs_url(base, prod["url"])) except Exception as e: logger.error(f"[{request_id}] ❌ JSON view parse error: {e}") # 3) Fallback: HTML scrape if not links: search_url = f"{base}/search?q={quote_plus(q)}" logger.info(f"[{request_id}] → Shopify HTML fallback: {search_url}") html = await fetch_text_local(client, search_url, referer=base) if html: logger.info(f"[{request_id}] ✅ HTML response: {len(html)} bytes") soup = BeautifulSoup(html, "html.parser") for a in soup.find_all("a", href=True): href = abs_url(base, a["href"]) if _match_product(href, s): links.append(href) if len(links) >= 10: break logger.info(f"[{request_id}] 📦 Scraped {len(links)} product links") else: logger.warning(f"[{request_id}] ⚠️ No HTML response") # ---------- WooCommerce ---------- elif s["platform"] == "woo": search_url = f"{base}/?s={quote_plus(q)}&post_type=product" logger.info(f"[{request_id}] → WooCommerce: {search_url}") html = await fetch_text_local(client, search_url, referer=base) if html: logger.info(f"[{request_id}] ✅ HTML response: {len(html)} bytes") soup = BeautifulSoup(html, "html.parser") for a in soup.find_all("a", href=True): href = abs_url(base, a["href"]) if _match_product(href, s): links.append(href) if len(links) >= 10: break logger.info(f"[{request_id}] 📦 Scraped {len(links)} product links") else: logger.warning(f"[{request_id}] ⚠️ No HTML response") # ---------- Custom ---------- else: # 🔧 FIX #2: Use search_url_template if defined search_template = s.get("search_url_template") if search_template: search_url = search_template.format(base=base, query=quote_plus(q)) logger.info(f"[{request_id}] → Custom (template): {search_url}") else: search_url = f"{base}/search?q={quote_plus(q)}" logger.info(f"[{request_id}] → Custom (generic): {search_url}") # Special Daraz handling if "daraz.pk" in base: logger.info(f"[{request_id}] 🔧 Daraz: using JS rendering") if USE_PLAYWRIGHT_DEFAULT: html = await fetch_js(search_url) else: html = await fetch_text_local(client, search_url, referer=base) else: html = await fetch_text_local(client, search_url, referer=base) if html: logger.info(f"[{request_id}] ✅ HTML response: {len(html)} bytes") soup = BeautifulSoup(html, "html.parser") for a in soup.find_all("a", href=True): href = abs_url(base, a["href"]) if _match_product(href, s): links.append(href) if len(links) >= 10: break logger.info(f"[{request_id}] 📦 Scraped {len(links)} product links") else: logger.warning(f"[{request_id}] ⚠️ No HTML response") logger.info(f"[{request_id}] ✅ {store_name}: collected {len(links)} links, returning top {min(len(links), 6)}") return links[:6] # Gather links from all stores logger.info(f"[{request_id}] 🚀 Starting parallel search across {len(STORES)} stores...") link_lists = await asyncio.gather(*[_links_for_store(s) for s in STORES]) # Flatten and deduplicate product_urls = [] seen = set() for links in link_lists: for url in links: if url not in seen: seen.add(url) product_urls.append(url) logger.info(f"[{request_id}] 📊 Total unique product URLs: {len(product_urls)}") logger.info(f"[{request_id}] 💾 Local cache entries: {len(local_cache)}") logger.info(f"[{request_id}] 💾 Global cache entries: {len(_cache)} (should not be used)") product_urls = product_urls[:limit * 3] # Parse product pages sem = asyncio.Semaphore(MAX_CONCURRENCY) logger.info(f"[{request_id}] 🔧 Parsing {min(len(product_urls), limit)} product pages...") async def _one(u: str): async with sem: try: store_config = guess_store(d(u)) or {"base": u, "platform": "custom"} return await parse_product( client, store_config, u, USE_PLAYWRIGHT_DEFAULT, fetch_func=fetch_text_local # ← CRITICAL: use per-request cache ) except Exception as e: logger.error(f"[{request_id}] ❌ Parse error for {u}: {e}") return [] parsed = await asyncio.gather(*[_one(u) for u in product_urls[:limit]], return_exceptions=True) logger.info(f"[{request_id}] ✅ Parsing complete") # Flatten + clean offers: List[Offer] = [] for r in parsed: if isinstance(r, Exception) or not r: continue offers.extend(r) logger.info(f"[{request_id}] 📦 Total offers extracted: {len(offers)}") # Build items items = [] for o in offers: if not o.title or o.price is None: continue items.append({ "source": STORE_DISPLAY_NAMES.get(o.source, o.source.replace(".pk", "").title()), "title": o.title, "price": o.price, "currency": o.currency or "PKR", "url": o.url, "image": o.image_url, "in_stock": o.in_stock, }) logger.info(f"[{request_id}] 🧹 After filtering: {len(items)} valid items") # De-duplicate by normalized title+source, keep cheapest keyd: Dict[Tuple[str, str], Dict[str, Any]] = {} for it in items: k = (it["source"], _norm(it["title"])) if k not in keyd or (it["price"] or 1e18) < (keyd[k]["price"] or 1e18): keyd[k] = it items = list(keyd.values()) logger.info(f"[{request_id}] 🔄 After deduplication: {len(items)} unique items") items.sort(key=lambda x: (x["price"] if x["price"] is not None else 1e18, x["title"].lower())) logger.info(f"[{request_id}] ✅ RETURNING {len(items[:limit])} offers for '{q}'") logger.info(f"[{request_id}] ========== REQUEST COMPLETE ==========\n") return {"query": q, "count": len(items[:limit]), "offers": items[:limit]}