Poshmark - Browse listings and extract title, brand, size, price, original price, URL

import gzip, json, math, re, socket, time, urllib.error, urllib.parse, urllib.request
 
UA = "poshmark-listing-export/1.0 (python-urllib; personal research script)"
BASE = "https://poshmark.com"
RETRYABLE = {429, 500, 502, 503, 504}
SITE_CAP_TOTAL = 5000   # more.total never exceeds this; has_next turns false at page 105
EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+")
PHONE = re.compile(
    r"(?<![\w+])(?:\+[\s.-]?|00)[1-9](?:[\s.()-]{0,2}\d){6,14}(?!\d)"   # international: +44 20 7946 0958, 0049 30 123456
    r"|(?<![\w+])0[1-9](?:[\s.()-]?\d){8,11}(?!\d)"                     # national trunk 0: 020 7946 0958, 07700 900123
    r"|(?<!\d)(?:1[\s.-]?)?\(?\d{3}\)?[\s.-]?\d{3}[\s.-]?\d{4}(?!\d)")  # North American: (555) 123-4567, 1-555-123-4567, 15551234567
ALLOWED_PATH = re.compile(r"(?:brand|category)/(?!\.\.?$)[^/?#\\]+")   # one segment under /brand or /category
 
def redact(text):
    """Titles are seller-written free text: mask any email address or phone number in them."""
    return PHONE.sub("[phone]", EMAIL.sub("[email]", text))
 
def read_body(r, deadline):
    """Read in chunks, each socket wait capped at the time left, so a server that trickles bytes cannot outlive the deadline."""
    sock, chunks = getattr(getattr(r.fp, "raw", None), "_sock", None), []
    while True:
        left = deadline - time.monotonic()
        if left <= 0:
            raise TimeoutError("deadline passed while reading the response body")
        if sock is not None:
            sock.settimeout(min(30, left))
        chunk = r.read1(65536)
        if not chunk:
            return b"".join(chunks)
        chunks.append(chunk)
 
def fetch(url, budget_s=120):
    """GET with capped exponential backoff under ONE deadline that also covers reading the body.
    Raises on 403/404 or when the budget is spent."""
    deadline, delay, attempt = time.monotonic() + min(budget_s, 120), 5, 0
    while True:
        attempt += 1
        remaining = deadline - time.monotonic()
        try:
            req = urllib.request.Request(url, headers={
                "User-Agent": UA, "Accept-Encoding": "gzip", "Accept-Language": "en-US,en;q=0.9"})
            with urllib.request.urlopen(req, timeout=max(0.1, min(30, remaining))) as r:
                body = read_body(r, deadline)
                if r.headers.get("Content-Encoding") == "gzip":
                    body = gzip.decompress(body)
                return r.geturl(), body.decode("utf-8")
        except urllib.error.HTTPError as e:
            if e.code not in RETRYABLE:
                raise
            err = e
        except (urllib.error.URLError, TimeoutError, socket.timeout, ConnectionError) as e:   # socket.timeout: Python < 3.10
            err = e
        if time.monotonic() + delay >= deadline:   # no sleep after the last attempt
            raise RuntimeError(f"giving up on {url} after {attempt} attempts: {err}")
        time.sleep(delay)
        delay *= 2
 
def browse_url(path, filters=None, page=1):
    """path: 'brand/<Brand>-<Dept>-<Category>' or 'category/<Dept>-<Category>' (both allowed by robots.txt).
    Anything else (e.g. 'search', which robots.txt disallows) raises before any request is made."""
    if not ALLOWED_PATH.fullmatch(path):
        raise ValueError(f"refusing {path!r}: only brand/<slug> and category/<slug> browse pages are allowed")
    q = list((filters or {}).items())
    if page > 1:
        q.append(("max_id", page))
    url = f"{BASE}/{urllib.parse.quote(path, safe='/-_')}"
    return url + ("?" + urllib.parse.urlencode(q, doseq=True) if q else "")
 
def parse_page(html):
    marker = "window.__INITIAL_STATE__="
    i = html.find(marker)
    if i < 0:
        raise RuntimeError("no __INITIAL_STATE__ in page (layout changed or not a listings page)")
    state = json.JSONDecoder().raw_decode(html, i + len(marker))[0]
    if state.get("showCaptcha"):
        raise RuntimeError("Poshmark served a captcha; stop and raise it to a human, do not retry")
    grid = next((v["gridData"] for k, v in state.items()
                 if k.startswith("$_") and isinstance(v, dict) and "gridData" in v), None)
    if grid is None:
        raise RuntimeError("no gridData module in __INITIAL_STATE__")
    def money(m):
        v = float((m or {}).get("val") or 0)
        return v if math.isfinite(v) and v > 0 else None   # 0.0 means "not set"; NaN/inf/negative rejected
    rows = []
    for it in grid.get("data") or []:
        bad = [f for f, ok in (("id", re.fullmatch(r"[0-9a-f]{24}", str(it.get("id") or ""))),
                               ("title", it.get("title")), ("price", money(it.get("price_amount"))))
               if not ok]
        if bad:
            raise RuntimeError(f"listing {it.get('id')!r} is missing {bad}; refusing the page")
        rows.append({
            "id": it["id"],
            "title": redact(it["title"]),
            "brand": it.get("brand"),
            "size": (it.get("size_obj") or {}).get("display_with_size_system") or it.get("size"),
            "price": money(it.get("price_amount")),
            "original_price": money(it.get("original_price_amount")),
            "currency": (it.get("price_amount") or {}).get("currency_code"),
            "department": (it.get("department") or {}).get("display"),
            "category": it.get("category"),
            "status": (it.get("inventory") or {}).get("status"),
            # ID-only permalink (301s to /listing/<Title-Slug>-<id>); the slug repeats the seller-written title.
            "url": f"{BASE}/listing/{it['id']}",
        })
    more = grid.get("more") or {}
    return rows, more.get("total"), bool(more.get("is_next_max_id_present"))
 
def same_url(a, b):
    """Same host, path and query (order and %-encoding ignored), so a redirect or a different page is never read."""
    pa, pb = urllib.parse.urlparse(a), urllib.parse.urlparse(b)
    q = lambda p: sorted(urllib.parse.parse_qsl(p.query, keep_blank_values=True))
    return (pa.netloc, urllib.parse.unquote(pa.path), q(pa)) == (pb.netloc, urllib.parse.unquote(pb.path), q(pb))
 
def scrape(path, filters=None, max_pages=3, pause_s=2.0):
    """Returns (rows, status). status is "complete", "truncated_max_pages" (more pages exist)
    or "truncated_site_cap" (Poshmark stopped paging at its ~5000-row cap; treat as a hard stop)."""
    seen, out = set(), []
    for page in range(1, max_pages + 1):
        url = browse_url(path, filters, page)
        final, html = fetch(url)
        if not same_url(final, url):
            raise RuntimeError(f"server sent us elsewhere: asked {url}, got {final}; "
                               "call scrape() again with that path if it is the page you want")
        rows, total, has_next = parse_page(html)
        if not rows and (has_next or total):
            raise RuntimeError(f"page {page} has 0 rows but total={total}, next={has_next}: {final}")
        new = [r for r in rows if r["id"] not in seen]
        seen.update(r["id"] for r in new)
        out.extend(new)
        print(f"page {page}: {len(rows)} rows, {len(new)} new, total={total}, next={has_next}")
        if not has_next:
            return out, ("truncated_site_cap" if (total or 0) >= SITE_CAP_TOTAL else "complete")
        if page == max_pages:
            return out, "truncated_max_pages"
        time.sleep(pause_s)
 
listings, status = scrape("brand/Levi's-Women-Jeans",
                          {"sort_by": "added_desc", "size[]": "28", "price[]": "25-50", "availability": "available"},
                          max_pages=3)
print(len(listings), "listings, status:", status)
for r in listings[:3]:
    print(json.dumps(r, ensure_ascii=False))
 
empty, status = scrape("brand/Zzqxnotabrand-Women-Jeans")
print(len(empty), "listings for an unknown brand, status:", status)
 
try:   # this slug 301s to a different canonical path, so scrape() refuses it and names the target
    scrape("brand/Lululemon_Athletica-Women-Pants", {"sort_by": "added_desc"}, max_pages=1)
except RuntimeError as e:
    print("refused:", e)
 
try:   # robots.txt disallows /search: refused locally, nothing is requested
    scrape("search", {"query": "levis"})
except ValueError as e:
    print("refused:", e)

import json, re, time, urllib.parse
 
ALLOWED_PATH = re.compile(r"/(?:brand|category)/(?!\.\.?$)[^/?#\\]+")   # robots.txt allows these; /search etc. are disallowed
 
EXTRACT = r"""(() => {
  if (document.querySelector('#px-captcha, iframe[src*="captcha"]')) return {error: 'challenge on page'};
  if (document.readyState !== 'complete') return {pending: 'readyState ' + document.readyState};
  const s = window.__INITIAL_STATE__;
  if (!s) return {pending: 'no __INITIAL_STATE__ yet'};
  if (s.showCaptcha) return {error: 'captcha'};
  const mod = Object.keys(s).find(k => k.startsWith('$_') && s[k] && s[k].gridData);
  if (!mod) return {pending: 'no gridData yet'};
  const g = s[mod].gridData, data = g.data || [];
  const money = m => { const v = parseFloat((m || {}).val || '0'); return Number.isFinite(v) && v > 0 ? v : null; };
  const bad = data.find(it => !/^[0-9a-f]{24}$/.test(it.id || '') || !it.title || !money(it.price_amount));
  if (bad) return {error: `listing ${bad.id} is missing id/title/price`};
  // Titles are seller-written free text: mask any email address or phone number.
  // Same patterns as the Python block: international (+44 ..., 0049 ...), national trunk 0, North American.
  const redact = t => t.replace(/[\w.+-]+@[\w-]+\.[\w.-]+/g, '[email]')
    .replace(/(?<![\w+])(?:\+[\s.-]?|00)[1-9](?:[\s.()-]{0,2}\d){6,14}(?!\d)|(?<![\w+])0[1-9](?:[\s.()-]?\d){8,11}(?!\d)|(?<!\d)(?:1[\s.-]?)?\(?\d{3}\)?[\s.-]?\d{3}[\s.-]?\d{4}(?!\d)/g, '[phone]');
  return {
    total: (g.more || {}).total, has_next: !!(g.more || {}).is_next_max_id_present,
    rows: data.map(it => ({
      id: it.id, title: redact(it.title), brand: it.brand,
      size: (it.size_obj || {}).display_with_size_system || it.size,
      price: money(it.price_amount), original_price: money(it.original_price_amount),
      // ID-only permalink (301s to /listing/<Title-Slug>-<id>); the slug repeats the seller-written title.
      url: location.origin + '/listing/' + it.id,
    })),
  };
})()"""
 
def same_url(a, b):
    """Same host, path and query (order and %-encoding ignored), so page N is never read as page N-1."""
    pa, pb = urllib.parse.urlparse(a), urllib.parse.urlparse(b)
    q = lambda p: sorted(urllib.parse.parse_qsl(p.query, keep_blank_values=True))
    return (pa.netloc, urllib.parse.unquote(pa.path), q(pa)) == (pb.netloc, urllib.parse.unquote(pb.path), q(pb))
 
def open_page(url, first=False, settle_s=20):
    u = urllib.parse.urlparse(url)
    if u.scheme != "https" or u.netloc != "poshmark.com" or not ALLOWED_PATH.fullmatch(urllib.parse.unquote(u.path)):
        raise ValueError(f"refusing {url}: only https://poshmark.com/brand/<slug> and /category/<slug> pages are allowed")
    new_tab(url) if first else goto_url(url)
    wait_for_load()
    want = urllib.parse.urlparse(url)
    deadline, href, data = time.monotonic() + settle_s, None, None
    while time.monotonic() < deadline:       # one deadline for the URL and state to settle
        href = js("location.href") or ""
        got = urllib.parse.urlparse(href)
        if got.netloc == want.netloc and urllib.parse.unquote(got.path) != urllib.parse.unquote(want.path):
            # pagination only changes the query, so a different path on this host is a server redirect
            raise RuntimeError(f"redirected: asked {url}, got {href}; open that URL instead if it is the page you want")
        if same_url(href, url):
            data = js(EXTRACT)
            if data and data.get("error"):
                raise RuntimeError(f"extraction failed on {href}: {data['error']}")
            if data and "rows" in data:
                return data
        time.sleep(0.5)
    raise RuntimeError(f"page did not settle on {url}: href={href} state={data and data.get('pending')}")
 
MAX_PAGES, SITE_CAP_TOTAL = 3, 5000
base = "https://poshmark.com/category/Men-Shoes?" + urllib.parse.urlencode(
    {"brand[]": "Nike", "sort_by": "added_desc"})
results, seen, status = [], set(), None
for page in range(1, MAX_PAGES + 1):
    url = base + (f"&max_id={page}" if page > 1 else "")
    data = open_page(url, first=(page == 1))
    new = [r for r in data["rows"] if r["id"] not in seen]
    seen.update(r["id"] for r in new)
    results.extend(new)
    print(f"page {page}: {len(data['rows'])} rows, {len(new)} new, next={data['has_next']}")
    if not data["rows"] and (data["has_next"] or data["total"]):
        raise RuntimeError(f"page {page} has 0 rows but total={data['total']}, next={data['has_next']}")
    if not data["has_next"]:
        status = "truncated_site_cap" if (data["total"] or 0) >= SITE_CAP_TOTAL else "complete"
        break
    if page == MAX_PAGES:
        status = "truncated_max_pages"
        break
    time.sleep(2)
print(len(results), "listings, status:", status)
print(json.dumps(results[:2], indent=1, ensure_ascii=False))
 
time.sleep(2)
try:   # this slug 301s to a different canonical path, so open_page() refuses it and names the target
    open_page("https://poshmark.com/brand/Lululemon_Athletica-Women-Pants?sort_by=added_desc")
except RuntimeError as e:
    print("refused:", e)
 
try:   # robots.txt disallows /search: refused before the tab navigates
    open_page("https://poshmark.com/search?query=nike")
except ValueError as e:
    print("refused:", e)