Costco - Search Results & Product Details

import json, time, urllib.parse
 
QUERY = "air fryer"   # any search term
MAX_PAGES = 3         # safety cap; the site shows 24 results per page. Reported when hit.
 
def page_url(n):
    # The site's own pager: page n is /s?keyword=...&currentPage=n (no pageSize or sortBy, which robots.txt disallows).
    params = {"keyword": QUERY}
    if n > 1:
        params["currentPage"] = n
    return "https://www.costco.com/s?" + urllib.parse.urlencode(params)
 
# Search results are not in the HTML. While loading, the page POSTs to its search API.
# Record each of those calls and its response (installed before the page loads), then read the one for this page.
RECORDER = """(() => {
  window.__costcoSearch = [];
  const orig = window.fetch;
  window.fetch = async function(input, init) {
    const url = typeof input === 'string' ? input : (input && input.url) || String(input);
    const isSearch = url.includes('/catalog/search/api/v1/search');
    let body = init && init.body;
    if (isSearch && input instanceof Request) body = await input.clone().text();   // the site's fetch wrapper passes a Request
    const res = await orig.apply(this, arguments);
    if (isSearch) {
      const entry = {status: res.status, body: typeof body === 'string' ? body : null, response: null};
      window.__costcoSearch.push(entry);
      res.clone().text().then(t => { entry.response = t; }).catch(() => { entry.response = ''; });
    }
    return res;
  };
})()"""
 
def js_retry(expr, tries=3):
    # js() reports a CDP timeout as RuntimeError("Runtime.evaluate timed out ..."); raw cdp raises TimeoutError.
    for i in range(tries):
        try:
            return js(expr)
        except TimeoutError:
            pass
        except RuntimeError as e:
            if "timed out" not in str(e):
                raise
        time.sleep(2)
    raise TimeoutError(f"js() timed out {tries} times: {expr[:80]}")
 
def request_of(call):
    try:
        b = json.loads(call.get("body") or "{}")
        return str(b.get("query", "")).strip().lower(), b.get("offset"), b.get("pageSize")
    except (ValueError, AttributeError):
        return None, None, None
 
def load_page(n):
    """Open search page n and return (page_size, response JSON) of the page's own search call for it."""
    goto_url(page_url(n))
    wait_for_load()
    seen = []
    for _ in range(30):
        url = js_retry("location.href") or ""
        q = urllib.parse.parse_qs(urllib.parse.urlsplit(url).query)
        if url.startswith("https://www.costco.com/s?") and q.get("keyword") == [QUERY]:
            seen = js_retry("(window.__costcoSearch || []).map((c, i) => ({i, status: c.status, body: c.body, done: c.response !== null}))") or []
            for c in seen:
                query, offset, size = request_of(c)
                size = size if isinstance(size, int) and size > 0 else 24
                if query != QUERY.lower() or offset != (n - 1) * size or not c["done"]:
                    continue
                if c["status"] != 200:
                    # 429/403 = rate limit or anti-bot challenge. Stop and back off; do not retry or work around it.
                    raise RuntimeError(f"search page {n}: the site's search call returned HTTP {c['status']}; stopping")
                return size, json.loads(js_retry(f"window.__costcoSearch[{c['i']}].response"))
        time.sleep(1)
    raise RuntimeError(f"search page {n}: no search response recorded within 30 s (calls seen: {[c['status'] for c in seen]})")
 
def attr(attrs, key):
    a = attrs.get(key) if isinstance(attrs.get(key), dict) else {}
    return (a.get("text") or []) + (a.get("numbers") or [])
 
def price_text(p):
    if not isinstance(p, dict):
        return None
    lo, hi = p.get("minPrice"), p.get("maxPrice")
    if not isinstance(lo, (str, int, float)) or lo in ("", 0):
        return None
    if not isinstance(hi, (str, int, float)) or hi in ("", 0):
        hi = lo
    return f"${lo}" if lo == hi else f"${lo} - ${hi}"
 
def parse(result, inventory):
    p = result.get("product") if isinstance(result.get("product"), dict) else {}
    attrs = p.get("attributes") if isinstance(p.get("attributes"), dict) else {}
    inv = inventory.get(result.get("id")) or {}
    rating = p.get("rating") if isinstance(p.get("rating"), dict) else {}
    avg, count = rating.get("averageRating"), rating.get("ratingCount")
    # Tiles with disp_price_in_cart_only == 1 show "Sign In for Details" instead of a price.
    hidden = 1.0 in attr(attrs, "disp_price_in_cart_only")
    variants = [v for v in p.get("variants") or [] if isinstance(v, dict)]
    return {
        "product_id": result.get("id"),
        "item_numbers": [v["id"] for v in variants if v.get("id")],   # "Item 1234567" on the product page
        "title": p.get("title"),
        "brand": (p.get("brands") or [None])[0],
        "price": None if hidden else price_text(inv.get("deliveryPrice")),
        "price_note": "Sign In for Details" if hidden else None,
        "rating": round(avg, 2) if isinstance(avg, (int, float)) and avg else None,
        "review_count": count if isinstance(count, int) else None,
        "url": p.get("uri"),
        "category": (p.get("categories") or [None])[-1],
        "pills": attr(attrs, "pills"),
        "delivery": inv.get("deliveryAvailability"),
    }
 
tab = new_tab()   # blank background tab; never brought to the front
try:
    cdp("Page.addScriptToEvaluateOnNewDocument", source=RECORDER)   # stays installed for every page this tab loads
    rows, seen, total, complete = [], set(), None, False
    for n in range(1, MAX_PAGES + 1):
        if n > 1:
            time.sleep(3)   # human-like pacing between pages
        page_size, data = load_page(n)
        sr = data.get("searchResult") or {}
        total = sr.get("totalSize")
        results = [r for r in sr.get("results") or [] if isinstance(r, dict)]
        inventory = {i.get("productId"): i for i in data.get("inventoryResponse") or [] if isinstance(i, dict)}
        for res in results:
            row = parse(res, inventory)
            if row["product_id"] is not None:   # dedupe by id; keep rows without one
                if row["product_id"] in seen:
                    continue
                seen.add(row["product_id"])
            rows.append(row)
        print(f"page {n}: {len(results)} results, {len(rows)} unique so far (totalSize {total})")
        if len(results) < page_size:   # only a short page ends the search
            complete = True
            break
    if not complete:
        print(f"MAX_PAGES cap hit: stopped after {MAX_PAGES} full pages, more results may remain (totalSize {total})")
 
    print("price shown:", sum(1 for r in rows if r["price"]),
          "| sign in for details:", sum(1 for r in rows if r["price_note"]),
          "| no online price:", sum(1 for r in rows if not r["price"] and not r["price_note"]))
    print(json.dumps(rows[:3], indent=1))
finally:
    close_tab(tab)

import email.utils, html, http.client, json, re, sys, time, urllib.error, urllib.request
from datetime import datetime, timezone
 
# Product URLs from the search rows (url field). Any https://www.costco.com/p/-/<slug>/<product_id>
# or <slug>.product.<product_id>.html URL works.
URLS = [
    "https://www.costco.com/p/-/t-fal-dual-window-infrared-air-fryer-73qt-dual-heating/4000430818",
    "https://www.costco.com/p/-/gourmia-9-qt-digital-air-fryer-with-ceramic-coating/4201032285",
    "https://www.costco.com/p/-/cuisinart-extra-large-26l-10-in-1-digital-air-fryer-oven/4000398962",
]
# A plain, truthful client UA works here; no browser UA needed.
HEADERS = {"User-Agent": f"Python-urllib/{sys.version_info[0]}.{sys.version_info[1]}", "Accept-Language": "en-US,en;q=0.9"}
RETRY_STATUS = {429, 500, 502, 503, 504}
MAX_WAIT = 120   # seconds; if the server asks to wait longer than this, stop instead of sleeping
 
def retry_after_seconds(value):
    """Retry-After is either seconds or an HTTP date."""
    if not value:
        return None
    try:
        return max(0.0, float(value))
    except ValueError:
        pass
    try:
        when = email.utils.parsedate_to_datetime(value)
    except (TypeError, ValueError):
        return None
    if when.tzinfo is None:
        when = when.replace(tzinfo=timezone.utc)
    return max(0.0, (when - datetime.now(timezone.utc)).total_seconds())
 
def get_page(url, tries=4):
    """(status, html or None). None only for 404. Retries 429/5xx/network errors with capped backoff, else raises."""
    last = None
    for attempt in range(tries):
        wait = None
        try:
            with urllib.request.urlopen(urllib.request.Request(url, headers=HEADERS), timeout=30) as r:
                return r.status, r.read().decode("utf-8", "replace")
        except urllib.error.HTTPError as e:
            if e.code == 404:
                return 404, None   # unknown or removed product
            if e.code not in RETRY_STATUS:
                raise RuntimeError(f"{url}: HTTP {e.code}; stopping")   # e.g. 403 = blocked: stop, do not work around it
            wait, last = retry_after_seconds(e.headers.get("Retry-After")), f"HTTP {e.code}"
        except (urllib.error.URLError, http.client.IncompleteRead, TimeoutError, ConnectionError) as e:
            last = repr(e)
        if attempt == tries - 1:
            break
        wait = wait if wait is not None else 2 ** (attempt + 1)
        if wait > MAX_WAIT:
            raise RuntimeError(f"{url}: {last}, server asked to wait {wait:.0f} s (over {MAX_WAIT} s); stopping")
        time.sleep(wait)
    raise RuntimeError(f"{url}: failed after {tries} tries ({last})")
 
def clean(fragment):
    return " ".join(html.unescape(re.sub(r"<[^>]+>", " ", fragment)).split())
 
def first_dict(v):
    if isinstance(v, list):
        v = next((x for x in v if isinstance(x, dict)), None)
    return v if isinstance(v, dict) else {}
 
def image_url(img):
    if isinstance(img, list):
        img = img[0] if img else None
    if isinstance(img, dict):   # schema.org ImageObject
        img = img.get("url") or img.get("contentUrl")
    return img if isinstance(img, str) else None
 
def product_details(url):
    status, page = get_page(url)
    if page is None:
        return {"url": url, "error": f"HTTP {status}"}
    ld = {}
    for block in re.findall(r'<script[^>]*application/ld\+json[^>]*>(.*?)</script>', page, re.S):
        try:
            data = json.loads(block)
        except ValueError:
            continue
        objs = data if isinstance(data, list) else (data.get("@graph") or [data]) if isinstance(data, dict) else []
        for obj in objs:
            if not isinstance(obj, dict):
                continue
            types = obj.get("@type") if isinstance(obj.get("@type"), list) else [obj.get("@type")]
            for t in types:
                if isinstance(t, str):
                    ld.setdefault(t, obj)
    prod = ld.get("Product")
    if not prod:
        return {"url": url, "error": "no Product JSON-LD (not a product page)"}
    offer = first_dict(prod.get("offers"))
    rating = first_dict(prod.get("aggregateRating"))
    brand = prod.get("brand")
    # Same rule as the search tiles: dispPriceInCartOnly 1 -> the page shows "Sign In for Price".
    # The first match is the main item (itemSequence 0); later ones are bundled or related items.
    flag = re.search(r'dispPriceInCartOnly\\":(\d)', page)
    hidden = bool(flag and flag.group(1) == "1")
    item_line = re.search(r'data-testid="Text_item-number">(.*?)</div>', page)   # "Item 1994092 | Model EY8338U0"
    model = re.search(r"Model\s+(\S+)", clean(item_line.group(1))) if item_line else None
    specs = {}
    for k, v in re.findall(r'<th[^>]*id="ProductSpecifications-th-row\d+-0"[^>]*>(.*?)</th><td[^>]*>(.*?)</td>', page):
        specs.setdefault(clean(k), []).append(clean(v))   # some keys repeat (e.g. two Capacity rows)
    crumbs = first_dict(ld.get("BreadcrumbList")).get("itemListElement") or []
    desc = prod.get("description")
    return {
        "url": prod.get("url") if isinstance(prod.get("url"), str) else url,
        "title": prod.get("name"),
        "item_number": prod.get("sku"),
        "model": model.group(1) if model else None,
        "brand": brand.get("name") if isinstance(brand, dict) else brand if isinstance(brand, str) else None,
        "price": None if hidden else offer.get("price"),
        "price_note": "Sign In for Price" if hidden else None,
        "currency": offer.get("priceCurrency"),
        "rating": rating.get("ratingValue"),
        "review_count": rating.get("ratingCount"),
        "breadcrumbs": [e.get("name") for e in crumbs if isinstance(e, dict)],
        "features": [clean(f) for f in re.findall(r'data-testid="Text_feature-\d+">(.*?)</div>', page)],
        "specs": {k: "; ".join(v) for k, v in specs.items()},
        "description": desc if isinstance(desc, str) else None,
        "image": image_url(prod.get("image")),
    }
 
for i, url in enumerate(URLS):
    if i:
        time.sleep(2)   # polite pacing
    d = product_details(url)
    if d.get("error"):
        print(url, d["error"])
        continue
    print(json.dumps({**d, "specs": dict(list(d["specs"].items())[:5]), "description": (d["description"] or "")[:120]}, indent=1))