Walmart Scraper: Prices and Products

# Walmart (and eBay): any URL
curl https://api.proxyfield.io/v1/scrape -H "Authorization: Bearer <api_key>" -H "Content-Type: application/json" \
  -d '{"url": "https://www.walmart.com/ip/example-product/123456789", "render": false, "country": "us", "output": "json"}'

# Amazon: an ASIN and a marketplace
curl https://api.proxyfield.io/v1/amazon/product -H "Authorization: Bearer <api_key>" -H "Content-Type: application/json" \
  -d '{"asin": "B0EXAMPLE1", "domain": "co.uk"}'

COMMON = ["marketplace", "item_id", "title", "price", "was_price", "currency", "availability",
          "condition", "shipping", "seller", "rating", "review_count", "url"]

def common(marketplace, row, domain="com"):
    """row: a parse_search() row (Walmart), a parse_results() row (eBay) or the Amazon API's `product`."""
    if marketplace == "amazon":
        row = {"item_id": row.get("asin"), "title": row.get("title"), "price": row.get("price"),
               "was_price": row.get("price_strikethrough"), "currency": row.get("currency"),
               "availability": row.get("stock"), "shipping": row.get("delivery"),
               "seller": (row.get("buybox") or {}).get("seller_name"), "rating": row.get("rating"),
               "review_count": row.get("reviews_count"),
               "url": f"https://www.amazon.{domain}/dp/{row.get('asin')}"}
    return {k: marketplace if k == "marketplace" else row.get(k) for k in COMMON}

{"marketplace": "walmart", "item_id": "988617972", "title": "Example Conical Burr Coffee Grinder", "price": 109.97, "was_price": 129.0, "currency": "USD", "availability": "IN_STOCK", "condition": null, "shipping": null, "seller": "Walmart.com", "rating": 4.4, "review_count": 696, "url": "https://www.walmart.com/ip/example/988617972"}
{"marketplace": "ebay", "item_id": "115998877665", "title": "Example Burr Coffee Grinder", "price": 58.99, "was_price": null, "currency": "USD", "availability": null, "condition": "Used", "shipping": "Free delivery", "seller": "techresale", "rating": null, "review_count": null, "url": "https://www.ebay.com/itm/115998877665"}
{"marketplace": "amazon", "item_id": "B0EXAMPLE7", "title": "Example Stainless Steel Burr Coffee Grinder", "price": 89.99, "was_price": 109.99, "currency": "USD", "availability": "In Stock", "condition": null, "shipping": [{"type": "FREE delivery", "date": "Wednesday, October 29"}], "seller": "Example Home Store", "rating": 4.4, "review_count": 5873, "url": "https://www.amazon.com/dp/B0EXAMPLE7"}

import json, re
import requests

# A rotating US residential IP on every request
proxy = "http://<login>-country-us:<password>@gate.proxyfield.io:7000"
url = "https://www.walmart.com/search?q=coffee+grinder&page=1"

def dig(obj, *path):
    for key in path:
        if isinstance(obj, list) and isinstance(key, int) and key < len(obj):
            obj = obj[key]
        elif isinstance(obj, dict):
            obj = obj.get(key)
        else:
            return None
    return obj

def parse_search(html):
    m = re.search(r'<script id="__NEXT_DATA__"[^>]*>(.*?)</script>', html, re.S)
    if not m:
        return None                                   # block or challenge page
    data = dig(json.loads(m.group(1)), "props", "pageProps", "initialData") or {}
    loc = dig(data, "pageMetadata", "location") or {}
    rows, seen = [], set()
    for stack in dig(data, "searchResult", "itemStacks") or []:
        for it in stack.get("items") or []:
            if it.get("__typename") != "Product" or it.get("usItemId") in seen:
                continue                              # ads, placeholders, repeats
            seen.add(it.get("usItemId"))
            lines = {l.get("lineType"): dig(l, "values", 0, "value")
                     for l in dig(it, "priceInfo", "priceDetails", "priceLines") or []}
            rows.append({
                "item_id": it.get("usItemId"),
                "title": it.get("name"),
                "price": it.get("price"),
                "was_price": float(lines["COMPARISON"]) if lines.get("COMPARISON") else None,
                "currency": dig(it, "priceInfo", "priceDetails", "currency"),
                "availability": dig(it, "availabilityStatusV2", "value"),
                "rating": it.get("averageRating"),
                "review_count": it.get("numberOfReviews"),
                "seller": it.get("sellerName"),
                "seller_id": it.get("sellerId"),
                "sponsored": bool(it.get("isSponsoredFlag")),
                "category": dig(it, "category", "categoryPath"),
                "image": dig(it, "imageInfo", "thumbnailUrl"),
                "url": "https://www.walmart.com" + (it.get("canonicalUrl") or "").split("?")[0],
                "store_id": loc.get("storeId"),
                "zip": loc.get("postalCode"),
            })
    return rows

def parse_product(html):                          # a walmart.com/ip/... page
    m = re.search(r'<script id="__NEXT_DATA__"[^>]*>(.*?)</script>', html, re.S)
    if not m:
        return None
    data = dig(json.loads(m.group(1)), "props", "pageProps", "initialData", "data") or {}
    p = data.get("product") or {}
    return {
        "item_id": p.get("usItemId"), "title": p.get("name"), "brand": p.get("brand"),
        "price": dig(p, "priceInfo", "currentPrice", "price"),
        "was_price": dig(p, "priceInfo", "wasPrice", "price"),
        "currency": dig(p, "priceInfo", "currentPrice", "currencyUnit"),
        "availability": p.get("availabilityStatus"),
        "seller": p.get("sellerName"), "seller_id": p.get("sellerId"),
        "rating": p.get("averageRating"),
        "review_count": dig(data, "reviews", "totalReviewCount"),
        "category": [c.get("name") for c in dig(p, "category", "path") or []],
        "images": [i.get("url") for i in dig(p, "imageInfo", "allImages") or []],
        "variants": [v.get("name") for v in p.get("variantCriteria") or []],
        "specifications": dig(data, "idml", "specifications") or [],
        "description": p.get("shortDescription"),
    }

resp = requests.get(url, proxies={"http": proxy, "https": proxy},
                    headers={"User-Agent": "Mozilla/5.0", "Accept-Language": "en-US"},
                    timeout=30)
rows = parse_search(resp.text) if resp.status_code == 200 else None
print("Blocked or challenge page: slow down and rotate the IP" if rows is None
      else json.dumps(rows[:2], indent=2))