Vinted - Catalog search listings (title, price, brand, size, condition, URL)

import json, math, re, time, urllib.parse
 
BASE = "https://www.vinted.com"
SITE_CAP_TOTAL = 960   # total_entries never exceeded 960 (10 pages x 96); page 11 returns 0 rows
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
    r"|(?<!\d)(?:1[\s.-]?)?\(?\d{3}\)?[\s.-]?\d{3}[\s.-]?\d{4}(?!\d)")  # North American: (555) 123-4567
# Label after the "<title>, " prefix. Brand and Size are left out when the seller did not set them.
LABEL = re.compile(r"(?:Brand: (?P<brand>.*?), )?Condition: (?P<condition>.*?), "
                   r"(?:Size: (?P<size>.*?), )?[\d.,]+ \$, [\d.,]+ \$")
 
EXTRACT = r"""(() => {
  if (document.querySelector('iframe[src*="captcha-delivery.com"]')) return {error: 'challenge on page'};
  if (document.readyState !== 'complete') return {pending: 'readyState ' + document.readyState};
  const P = 'self.__next_f.push(';
  const flight = [...document.scripts].map(s => s.textContent).filter(t => t.startsWith(P)).map(t => {
    try { const a = JSON.parse(t.slice(P.length, t.lastIndexOf(')'))); return typeof a[1] === 'string' ? a[1] : ''; }
    catch (e) { return ''; }
  }).join('');
  const key = '"items":{"items":[';
  const i = flight.indexOf(key);
  if (i < 0) return {pending: 'no catalog items in the page payload yet'};
  let j = i + key.length - 1, depth = 0, inStr = false, esc = false;   // find the matching ]
  for (; j < flight.length; j++) {
    const c = flight[j];
    if (inStr) { if (esc) esc = false; else if (c === '\\') esc = true; else if (c === '"') inStr = false; continue; }
    if (c === '"') inStr = true; else if (c === '[' || c === '{') depth++;
    else if (c === ']' || c === '}') { if (--depth === 0) break; }
  }
  const items = JSON.parse(flight.slice(i + key.length - 1, j + 1));
  const pg = flight.slice(j + 1, j + 400).match(/^,"pagination":(\{[^}]*\})/);
  return {
    pagination: pg ? JSON.parse(pg[1]) : null,
    rows: items.map(x => { const p = x.productItem || {}, b = p.itemBox || {};
      return {id: p.id, title: p.title, label: b.accessibilityLabel, second_line: b.secondLine,
              price: p.price, discounted: p.priceWithDiscount, total: p.totalItemPrice, promoted: !!p.isPromoted}; }),
  };
})()"""
 
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 money(m):
    try:
        v = float((m or {}).get("amount") or 0)
    except ValueError:
        return None
    return v if math.isfinite(v) and v > 0 else None
 
def parse_row(r):
    if not isinstance(r["id"], int) or not r["title"] or money(r["price"]) is None:
        raise RuntimeError(f"row {r['id']!r} is missing id/title/price; refusing the page")
    prefix = r["title"] + ", "
    m = LABEL.fullmatch(r["label"][len(prefix):]) if (r["label"] or "").startswith(prefix) else None
    if not m:
        raise RuntimeError(f"row {r['id']}: unexpected label {r['label']!r}")
    size, condition = m["size"], m["condition"]
    if r["second_line"] != " · ".join(x for x in (size, condition) if x):   # what the tile shows
        raise RuntimeError(f"row {r['id']}: label and tile disagree: {r['label']!r} vs {r['second_line']!r}")
    return {
        "id": r["id"],
        "title": redact(r["title"]),
        "brand": m["brand"],
        "size": size,
        "condition": condition,
        "price": money(r["price"]),
        "discounted_price": money(r["discounted"]),
        "total_price": money(r["total"]),           # price + buyer protection fee, as the tile shows
        "currency": r["price"].get("currencyCode"),
        "promoted": r["promoted"],
        # ID-only permalink; it redirects to /items/<id>-<title-slug>, and the slug repeats the seller-written title.
        "url": f"{BASE}/items/{r['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))
 
_tab = {"open": False}
 
def open_page(url, settle_s=25):
    u = urllib.parse.urlparse(url)
    if u.scheme != "https" or u.netloc != "www.vinted.com" or u.path != "/catalog":
        raise ValueError(f"refusing {url}: only https://www.vinted.com/catalog?... pages are used")
    goto_url(url) if _tab["open"] else new_tab(url)
    _tab["open"] = True
    wait_for_load()
    deadline, href, data = time.monotonic() + settle_s, None, None
    while time.monotonic() < deadline:   # one deadline for the URL and the payload to settle
        href = js("location.href") or ""
        if same_url(href, url):
            data = js(EXTRACT)
            if data and data.get("error"):
                raise RuntimeError(f"{data['error']} on {href}: stop, do not retry or change the request")
            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')}")
 
def search(query, filters=None, max_pages=3, pause_s=3.0):
    """Returns (rows, status). status is "complete", "truncated_max_pages" (more pages exist)
    or "truncated_site_cap" (Vinted stops at 10 pages / 960 rows; treat as a hard stop)."""
    seen, out = set(), []
    for page in range(1, max_pages + 1):
        q = [("search_text", query)] + list((filters or {}).items()) + ([("page", page)] if page > 1 else [])
        data = open_page(f"{BASE}/catalog?" + urllib.parse.urlencode(q, doseq=True))
        pg = data["pagination"] or {}
        rows = [parse_row(r) for r in data["rows"]]
        if not rows and pg.get("total_entries"):
            raise RuntimeError(f"page {page} has 0 rows but total_entries={pg.get('total_entries')}")
        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_entries={pg.get('total_entries')}, "
              f"total_pages={pg.get('total_pages')}")
        if page >= (pg.get("total_pages") or 0):
            return out, ("truncated_site_cap" if (pg.get("total_entries") or 0) >= SITE_CAP_TOTAL else "complete")
        if page == max_pages:
            return out, "truncated_max_pages"
        time.sleep(pause_s)
 
listings, status = search("nike", max_pages=3)
print(len(listings), "listings, status:", status)
for r in listings[:2]:
    print(json.dumps(r, ensure_ascii=False))
 
time.sleep(3)   # filters exactly as the site's own filter UI writes them into the URL
listings, status = search("levis jeans", {"order": "newest_first", "status_ids[]": "6",
                                          "price_to": "30", "currency": "USD"}, max_pages=1)
print(len(listings), "listings, status:", status,
      "| conditions:", sorted({r["condition"] for r in listings}),
      "| max price:", max(r["price"] for r in listings))
 
time.sleep(3)
listings, status = search("vintage dress", max_pages=1)
no_brand = [r for r in listings if r["brand"] is None]
print(len(listings), "listings,", len(no_brand), "without a brand")
print(json.dumps(no_brand[0] if no_brand else None, ensure_ascii=False))
 
time.sleep(3)
empty, status = search("zzqxqvnotathing")
print(len(empty), "listings for a nonsense query, status:", status)
 
try:   # robots.txt disallows /member/: refused before the tab navigates
    open_page("https://www.vinted.com/member/12345")
except ValueError as e:
    print("refused:", e)