Target - Category Listings & Product Details

import base64, html, json, re, time
 
CATEGORY_URL = "https://www.target.com/c/air-fryers-kitchen-appliances-dining/-/N-ncrpx"   # any /c/<slug>/-/N-<id> page
MAX_PAGES = 3   # 24 organic results per page
 
if not re.fullmatch(r"https://www\.target\.com/c/[^?#]+/-/N-[0-9A-Za-z]+", CATEGORY_URL):
    raise SystemExit("CATEGORY_URL must be https://www.target.com/c/<slug>/-/N-<id> with no query string")
 
LISTING = "/redsky_aggregations/v1/web/plp_search_v2"
SUMMARY = "/redsky_aggregations/v1/web/product_summary_with_fulfillment_v1"
 
def blocked(status, what="listing"):
    return (f"Target answered the category page's own {what} request with HTTP {status} "
            "(PerimeterX returns 435). Stop and back off; do not retry or try to solve the captcha.")
 
def body(session, request_id):
    r = cdp("Network.getResponseBody", session_id=session, requestId=request_id)
    return json.loads(base64.b64decode(r["body"]) if r.get("base64Encoded") else r["body"])
 
def navigate(url, timeout=15):
    """Load url in the attached tab and confirm the tab is really there before reading."""
    before = js("location.href")
    goto_url(url)
    deadline = time.time() + timeout
    while (href := js("location.href")) != url:
        if time.time() > deadline:
            raise RuntimeError(f"asked for {url} (tab was on {before}), tab is on {href}")
        time.sleep(0.2)
 
def read_page(session, timeout=30):
    """Read the JSON the page itself fetched while loading (no extra requests)."""
    reqs, listing, ful, start = {}, None, {}, time.time()
    while time.time() < start + timeout:
        for e in drain_events():
            if e.get("session_id") != session:   # events from other attached tabs
                continue
            m, p = e.get("method"), e.get("params") or {}
            if m == "Network.requestWillBeSent":
                url = p["request"]["url"]
                kind = "listing" if LISTING in url else "summary" if SUMMARY in url else None
                if kind:
                    reqs[p["requestId"]] = {"kind": kind}
            elif p.get("requestId") not in reqs:
                continue
            elif m == "Network.responseReceived":
                reqs[p["requestId"]]["status"] = p["response"]["status"]
            elif m == "Network.loadingFailed":
                raise RuntimeError(f"{reqs[p['requestId']]['kind']} request failed: {p.get('errorText')}")
            elif m == "Network.loadingFinished":
                r = reqs[p["requestId"]]
                if r.get("status") != 200:
                    if r["kind"] == "listing":
                        raise SystemExit(blocked(r.get("status")))
                    if r.get("status") == 435:
                        raise SystemExit(blocked(435, "fulfillment"))
                    raise RuntimeError(f"fulfillment request HTTP {r.get('status')}")
                data = body(session, p["requestId"])["data"]
                if r["kind"] == "listing":
                    listing = data["search"]
                else:
                    for s in data.get("product_summaries") or []:
                        ful[s["tcin"]] = s.get("fulfillment") or {}
        if listing is not None:
            missing = {p["tcin"] for p in listing.get("products") or []} - ful.keys()
            if not missing:
                return listing, ful
        time.sleep(0.2)
    if listing is None:
        raise RuntimeError(f"no {LISTING} response within {timeout} s, check the page: {page_info()}")
    raise RuntimeError(f"no fulfillment for {len(missing)} of {len(listing['products'])} TCINs within {timeout} s; "
                       "department landing pages (no 'page 1 of N' pager) do this, use a subcategory")
 
def parse(p, f):
    item = p.get("item") or {}
    parent = p.get("parent") or {}
    price = p.get("price") or parent.get("price") or {}
    stats = (p.get("ratings_and_reviews") or parent.get("ratings_and_reviews") or {}).get("statistics") or {}
    rating = stats.get("rating") or {}
    stores = f.get("store_options") or [{}]
    return {
        "tcin": p.get("tcin"),
        "title": html.unescape((item.get("product_description") or {}).get("title") or ""),
        "brand": (item.get("primary_brand") or {}).get("name"),
        "price": price.get("formatted_current_price"),   # "$59.99" or "See price in cart"
        "price_value": price.get("current_retail"),      # float, use .get: not every price object has it
        "reg_price": price.get("reg_retail"),
        "rating": rating.get("average"),                 # None for unrated items
        "rating_count": rating.get("count"),
        "url": (item.get("enrichment") or {}).get("buy_url"),
        "ship_status": (f.get("shipping_options") or {}).get("availability_status"),
        "pickup_status": (stores[0].get("order_pickup") or {}).get("availability_status"),
        "sold_out": f.get("sold_out"),
        "sponsored": bool(p.get("is_sponsored_sku")),
    }
 
def is_blank(url):   # the same blank-tab cases browser-harness's own new_tab() reuses
    url = url or ""
    return (url in ("", "about:blank", "data:text/html,") or url.startswith("about:blank#")
            or url.startswith(("chrome://newtab", "chrome://new-tab-page", "edge://newtab", "about:newtab")))
 
try:
    cur = current_tab()
except Exception:
    cur = {}
rows, seen, meta, tab = [], set(), {}, None
try:
    if cur.get("targetId") and is_blank(cur.get("url")):
        tab = cur.get("targetId")   # reuse the blank tab
    else:   # what new_tab() does, but the id is known before attaching, so finally can close it
        tab = cdp("Target.createTarget", url="about:blank", background=True)["targetId"]
    session = switch_tab(tab)     # its CDP session id; read_page ignores events from any other tab
    for page in range(MAX_PAGES):
        if page:
            time.sleep(3)   # polite pacing between page loads
        drain_events()
        navigate(f"{CATEGORY_URL}?Nao={page * 24}" if page else CATEGORY_URL)   # the page's own pagination URL
        listing, ful = read_page(session)
        meta = (listing.get("search_response") or {}).get("metadata") or {}
        if meta.get("offset") != page * 24:
            raise RuntimeError(f"asked for offset {page * 24}, page returned {meta.get('offset')}")
        products = listing.get("products") or []
        if not products:
            break
        for p in products:
            row = parse(p, ful[p["tcin"]])
            if row["sponsored"] or row["tcin"] in seen:   # sponsored slots and cross-page repeats
                continue
            seen.add(row["tcin"])
            rows.append(row)
        print(f"page {meta.get('current_page')}/{meta.get('total_pages')}: {len(products)} products, {len(rows)} unique so far")
        if meta.get("current_page", 0) >= (meta.get("total_pages") or 0):
            break
finally:
    if tab:
        close_tab(tab)
 
print("total_results:", meta.get("total_results"))
print(json.dumps(rows[:2], indent=1))

import html, json, re
 
TCIN = "89110863"   # from the category rows above, or the A-<digits> part of any /p/ URL
TCIN = TCIN.strip().removeprefix("A-")   # accept "A-89110863" too
 
page = http_get(f"https://www.target.com/p/-/A-{TCIN}", headers={
    "User-Agent": "browser-harness/1.0 (target domain skill)",
    "Accept": "text/html,application/xhtml+xml",
    "Accept-Language": "en-US,en;q=0.9",
})
m = re.search(r'<script id="__NEXT_DATA__"[^>]*>(.*?)</script>', page, re.S)
if not m:
    raise SystemExit("no __NEXT_DATA__ (block page or layout change)")
queries = (json.loads(m.group(1)).get("props") or {}).get("dehydratedState", {}).get("queries") or []
layout = next((q["state"]["data"]["data"] for q in queries
               if str(q.get("queryKey", [""])[0]).endswith("fetch-cdui-layout-v1")), None)
core = next((mod["module_data"]["data"]["product"] for mod in (layout or {}).get("data_source_modules") or []
             if mod.get("module_type") == "ProductDetailWebDatasourceCore"), None)
if not core:
    raise SystemExit("no ProductDetailWebDatasourceCore module (block page, bad TCIN or layout change)")
 
item = core.get("item") or {}
desc = item.get("product_description") or {}
stats = (core.get("ratings_and_reviews") or {}).get("statistics") or {}
rating = stats.get("rating") or {}
images = (item.get("enrichment") or {}).get("image_info") or {}
detail = {
    "tcin": core.get("tcin"),
    "title": html.unescape(desc.get("title") or ""),
    "brand": (item.get("primary_brand") or {}).get("name"),
    "dpci": item.get("dpci"),
    "upc": (item.get("primary_barcode") or None),
    "url": (item.get("enrichment") or {}).get("buy_url"),
    "category_path": [b["name"] for b in (core.get("category") or {}).get("breadcrumbs") or []][1:],
    "rating": rating.get("average"),
    "rating_count": rating.get("count"),
    "review_count": stats.get("review_count"),
    "question_count": stats.get("question_count"),
    "rating_distribution": rating.get("distribution"),
    "highlights": [html.unescape(re.sub("<[^>]+>", "", b)) for b in (desc.get("soft_bullets") or {}).get("bullets") or []],
    "specs": [html.unescape(re.sub("<[^>]+>", "", b)) for b in desc.get("bullet_descriptions") or []],
    "description": html.unescape(re.sub("<[^>]+>", " ", desc.get("downstream_description") or "")).strip(),
    "package_dimensions": item.get("package_dimensions"),
    "primary_image": (images.get("primary_image") or {}).get("url"),
    "image_count": int(bool((images.get("primary_image") or {}).get("url"))) + len(images.get("alternate_images") or []),
}
print(json.dumps(detail, indent=1)[:2500])

import json, time
 
TCIN = "89110863".strip().removeprefix("A-")
ALLOW_FOREGROUND = False   # opt-in: True lets this block bring the tab to the front (visible tab switch)
 
URL = f"https://www.target.com/p/-/A-{TCIN}"
 
# what new_tab() does, but the id is known before attaching, so finally can close it
tab = cdp("Target.createTarget", url="about:blank", background=True)["targetId"]
try:
    switch_tab(tab)
    before = js("location.href")
    goto_url(URL)
    deadline = time.time() + 15
    while (href := js("location.href")) != URL:   # confirm the tab really is on the requested page before reading it
        if time.time() > deadline:
            raise SystemExit(f"asked for {URL} (tab was on {before}), tab is on {href}")
        time.sleep(0.2)
    wait_for_load()
 
    # Price and fulfillment are deferred modules, filled after load by
    # www.target.com/cdui_orchestrations/v1/pages/pdp/deferred_enrichment/modules
    # They stay a grey placeholder while the tab is hidden (document.visibilityState == "hidden").
    live = {}
    for i in range(20):
        live = js("""(() => {
          const t = s => (document.querySelector(s)?.innerText || '').trim();
          return {
            visibility: document.visibilityState,
            title: t('[data-test="product-title"]'),
            price_block: t('[data-test="module-product-detail-price-v2"]'),
            fulfillment: t('[data-test="module-product-detail-fulfillment-v1"]'),
            rating_block: t('[data-test="ratingFeedbackContainer"]'),
          };
        })()""")
        if live["price_block"] and live["fulfillment"]:
            break
        if i == 5 and live["visibility"] == "hidden":
            if not ALLOW_FOREGROUND:
                raise SystemExit("tab is hidden and the price/fulfillment modules did not render; "
                                 "set ALLOW_FOREGROUND = True to bring it to the front")
            activate_tab(current_tab())
        time.sleep(1)
    else:
        raise SystemExit(f"price/fulfillment modules did not render in 20 s: {json.dumps(live)}")
    print(json.dumps(live, indent=1))
finally:
    close_tab(tab)