TCGplayer - Card Search, Market Price and Listings

import base64, json, re, time, urllib.parse
 
CRAWL_DELAY_S = 10        # robots.txt: Crawl-Delay: 10
SEARCH_API = "mp-search-api.tcgplayer.com/v1/search/request"
 
def search_url(line, q, page=1):
    """The same URL shape the site's own 'Shop All' links and pager use."""
    if not re.fullmatch(r"[a-z0-9]+(?:-[a-z0-9]+)*", line) or not q.strip():
        raise ValueError("line must be a productLineUrlName like 'magic' and q must be non-empty")
    qs = urllib.parse.urlencode({"productLineName": line, "q": q, "view": "grid", "page": page})
    url = f"https://www.tcgplayer.com/search/{line}/product?{qs}"
    if "seller=" in url:   # robots.txt: Disallow: /search/*/product?seller=*
        raise ValueError("refusing a seller-filtered search URL")
    return url
 
def challenge_check():
    t = (js("document.title") or "").lower()
    body = (js("document.body ? document.body.innerText.slice(0, 400) : ''") or "").lower()
    if any(s in t + body for s in ("just a moment", "access denied", "attention required", "verify you are human")):
        raise SystemExit(f"challenge page on {js('location.href')}: stop and back off, do not retry")
 
def read_search(session, want_q, want_from, timeout=30):
    """Return the JSON of the search request the page itself sent (no extra requests are made)."""
    reqs, start = {}, time.time()
    while time.time() < start + timeout:
        for e in drain_events():
            if e.get("session_id") != session:
                continue
            m, p = e.get("method"), e.get("params") or {}
            if m == "Network.requestWillBeSent":
                r = p["request"]
                q = urllib.parse.parse_qs(urllib.parse.urlparse(r["url"]).query).get("q", [""])[0]
                if SEARCH_API in r["url"] and r["method"] == "POST" and q == want_q:
                    reqs[p["requestId"]] = {"post": json.loads(r.get("postData") or "{}")}
            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"search request failed: {p.get('errorText')}")
            elif m == "Network.loadingFinished":
                r = reqs[p["requestId"]]
                if r.get("status") != 200:
                    raise SystemExit(f"search API answered HTTP {r.get('status')}: stop and back off, do not retry")
                if r["post"].get("from") != want_from:
                    continue   # an earlier page's request still finishing
                b = cdp("Network.getResponseBody", session_id=session, requestId=p["requestId"])
                data = json.loads(base64.b64decode(b["body"]) if b.get("base64Encoded") else b["body"])
                if data.get("errors"):
                    raise RuntimeError(f"search API errors: {data['errors']}")
                return data["results"][0]
        time.sleep(0.2)
    challenge_check()
    raise RuntimeError(f"no {SEARCH_API} response with from={want_from} within {timeout} s: {page_info()}")
 
def row(it):
    ca = it.get("customAttributes") or {}
    pid = int(it["productId"])
    return {
        "product_id": pid,
        "name": it.get("productName"),
        "set": it.get("setName"),
        "set_code": it.get("setCode"),
        "rarity": it.get("rarityName"),
        "number": ca.get("number"),
        "market_price": it.get("marketPrice"),
        "lowest_price": it.get("lowestPrice"),
        "lowest_price_with_shipping": it.get("lowestPriceWithShipping"),
        "listings": int(it.get("totalListings") or 0),
        "sealed": it.get("sealed"),
        "url": f"https://www.tcgplayer.com/product/{pid}",
    }
 
def search(session, line, q, max_pages=2):
    """Returns (rows, total, status); status is "complete" or "truncated_max_pages"."""
    rows, seen = [], set()
    for page in range(1, max_pages + 1):
        if page > 1:
            time.sleep(CRAWL_DELAY_S)
        drain_events()
        goto_url(search_url(line, q, page))
        res = read_search(session, q, want_from=(page - 1) * 24)
        total, items = int(res["totalResults"]), res.get("results") or []
        new = [row(it) for it in items if int(it["productId"]) not in seen]
        seen.update(r["product_id"] for r in new)
        rows.extend(new)
        print(f"{line} {q!r} page {page}: {len(items)} products, {len(new)} new, total={total}")
        if page * 24 >= total or not items:
            return rows, total, "complete"
    return rows, total, "truncated_max_pages"
 
tab = cdp("Target.createTarget", url="about:blank", background=True)["targetId"]
session = switch_tab(tab)
try:
    rows, total, status = search(session, "magic", "lightning bolt", max_pages=2)
    print(len(rows), "products of", total, "status:", status)
    for r in rows[:3]:
        print(json.dumps(r, ensure_ascii=False))
    time.sleep(CRAWL_DELAY_S)
    rows, total, status = search(session, "pokemon", "charizard ex", max_pages=1)
    print(len(rows), "products of", total, "status:", status)
    print(json.dumps(rows[0], ensure_ascii=False))
    time.sleep(CRAWL_DELAY_S)
    rows, total, status = search(session, "magic", "zzqx notacard")
    print(len(rows), "products of", total, "status:", status)
finally:
    close_tab(tab)
 
try:   # robots.txt disallows /search/*/product?seller=*; a bad line slug is refused the same way
    search_url("magic/product?seller=1", "bolt")
except ValueError as e:
    print("refused:", e)

import base64, json, re, time
 
PRODUCT_IDS = [1479, 719091, 1032]   # product_id values from block 1
CRAWL_DELAY_S = 10             # robots.txt: Crawl-Delay: 10
 
def challenge_check():
    t = (js("document.title") or "").lower()
    body = (js("document.body ? document.body.innerText.slice(0, 400) : ''") or "").lower()
    if any(s in t + body for s in ("just a moment", "access denied", "attention required", "verify you are human")):
        raise SystemExit(f"challenge page on {js('location.href')}: stop and back off, do not retry")
 
def read_product(session, pid, timeout=30):
    """Collect the JSON the product page itself loads: details, its cheapest-listings call, latest sales."""
    pats = {
        "details": re.compile(rf"mp-search-api\.tcgplayer\.com/v2/product/{pid}/details\b"),
        "listings": re.compile(rf"mp-search-api\.tcgplayer\.com/v1/product/{pid}/listings\b"),
        "sales": re.compile(rf"mpapi\.tcgplayer\.com/v2/product/{pid}/latestsales\b"),
    }
    reqs, got, start = {}, {}, time.time()
    while time.time() < start + timeout:
        for e in drain_events():
            if e.get("session_id") != session:
                continue
            m, p = e.get("method"), e.get("params") or {}
            if m == "Network.requestWillBeSent":
                r = p["request"]
                kind = next((k for k, rx in pats.items() if rx.search(r["url"])), None)
                if kind and r["method"] in ("GET", "POST"):
                    reqs[p["requestId"]] = {"kind": kind, "post": json.loads(r.get("postData") or "{}")}
            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:
                    raise SystemExit(f"{r['kind']} answered HTTP {r.get('status')}: stop and back off, do not retry")
                # the page sends listings twice (size 0 for filter counts, then size 10) and
                # latestsales twice (limit 1, then limit 25); keep the larger one
                size = r["post"].get("size", r["post"].get("limit", 0))
                if r["kind"] in got and got[r["kind"]][0] >= size:
                    continue
                b = cdp("Network.getResponseBody", session_id=session, requestId=p["requestId"])
                got[r["kind"]] = (size, json.loads(base64.b64decode(b["body"]) if b.get("base64Encoded") else b["body"]))
        if "details" in got and got.get("listings", (0,))[0] > 0 and got.get("sales", (0,))[0] > 1:
            break
        time.sleep(0.2)
    if "details" not in got:
        challenge_check()
        raise RuntimeError(f"no details response for product {pid} within {timeout} s: {page_info()}")
    return {k: v[1] for k, v in got.items()}
 
def parse(pid, raw):
    d = raw["details"]
    if int(d.get("productId") or 0) != pid:
        raise RuntimeError(f"asked for product {pid}, page loaded {d.get('productId')}")
    lres = ((raw.get("listings") or {}).get("results") or [{}])[0]
    sales = raw.get("sales") or {}
    return {
        "product_id": pid,
        "name": d.get("productName"),
        "set": d.get("setName"),
        "line": d.get("productLineName"),
        "rarity": d.get("rarityName"),
        "market_price": d.get("marketPrice"),
        "median_price": d.get("medianPrice"),
        "lowest_price": d.get("lowestPrice"),
        "lowest_price_with_shipping": d.get("lowestPriceWithShipping"),
        "listings_total": lres.get("totalResults"),
        "skus": [(s["sku"], s["condition"], s["variant"], s["language"]) for s in d.get("skus") or []],
        "cheapest_listings": [{
            "condition": l.get("condition"), "printing": l.get("printing"), "language": l.get("language"),
            "price": l.get("price"), "shipping": l.get("shippingPrice"), "quantity": int(l.get("quantity") or 0),
            "seller_rating": l.get("sellerRating"), "seller_sales": l.get("sellerSales"),
            "direct": l.get("directListing"), "listing_type": l.get("listingType"),
        } for l in lres.get("results") or []],
        "latest_sales": [{
            "date": s.get("orderDate"), "condition": s.get("condition"), "variant": s.get("variant"),
            "price": s.get("purchasePrice"), "shipping": s.get("shippingPrice"), "quantity": s.get("quantity"),
        } for s in sales.get("data") or []],
        "sales_meta": {k: sales.get(k) for k in ("totalResults", "nextPage")},
        "url": f"https://www.tcgplayer.com/product/{pid}",
    }
 
tab = cdp("Target.createTarget", url="about:blank", background=True)["targetId"]
session = switch_tab(tab)
out = []
try:
    for i, pid in enumerate(PRODUCT_IDS):
        if i:
            time.sleep(CRAWL_DELAY_S)
        drain_events()
        goto_url(f"https://www.tcgplayer.com/product/{int(pid)}")
        out.append(parse(int(pid), read_product(session, int(pid))))
finally:
    close_tab(tab)
 
for p in out:
    print(json.dumps({k: v for k, v in p.items() if k not in ("cheapest_listings", "latest_sales", "skus", "sales_meta")}, ensure_ascii=False))
    print(f"  skus={len(p['skus'])} cheapest_listings={len(p['cheapest_listings'])} "
          f"latest_sales={len(p['latest_sales'])} sales_meta={p['sales_meta']}")
    print("  cheapest:", json.dumps(p["cheapest_listings"][:2]))
    print("  latest sale:", json.dumps(p["latest_sales"][:1]))