AliExpress - Product Search & Product Page Scraping

import email.utils, gzip, http.client, json, re, time, urllib.error, urllib.parse, urllib.request
 
# Truthful client User-Agent; it returned the full search page on 2026-10-23.
HEADERS = {
    "User-Agent": "browser-harness-skill/1.0 (+https://github.com/browser-use/browser-harness)",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.9",
    "Accept-Encoding": "gzip",
}
SITE_MAX_PAGES = 60                       # AliExpress serves at most 60 pages per query
MAX_WAIT = 120                            # longest Retry-After / backoff sleep; a longer ask stops the run
 
 
class SearchBlocked(RuntimeError):
    """A search page without usable data (captcha or layout change). Stop; do not retry in a loop."""
 
 
def search_url(query, page=1, sort=None):
    # Non-Latin queries give an empty slug ("wholesale-.html"); that URL also returned results.
    slug = re.sub(r"[^a-z0-9]+", "-", query.lower()).strip("-")
    params = {"SearchText": query, "page": page, **({"sortType": sort} if sort else {})}
    return f"https://www.aliexpress.com/w/wholesale-{slug}.html?{urllib.parse.urlencode(params)}"   # safe="": "/" -> %2F
 
 
def retry_after_seconds(value):
    """Retry-After is either delta-seconds or an HTTP-date. -> seconds or None."""
    value = (value or "").strip()
    if value.isdigit():
        return int(value)
    try:
        return max(0.0, email.utils.parsedate_to_datetime(value).timestamp() - time.time())
    except (TypeError, ValueError, IndexError):
        return None
 
 
def fetch(url, attempts=4, timeout=30):
    """GET with capped backoff on 429, 5xx and network errors. Raises if every attempt fails
    or the server asks to wait longer than MAX_WAIT."""
    delay = 5
    for attempt in range(1, attempts + 1):
        try:
            with urllib.request.urlopen(urllib.request.Request(url, headers=HEADERS), timeout=timeout) as r:
                body = r.read()
                if r.headers.get("Content-Encoding") == "gzip":
                    body = gzip.decompress(body)
                return body.decode("utf-8", "replace")
        except urllib.error.HTTPError as e:   # subclass of URLError, so it is caught first
            if e.code != 429 and e.code < 500:
                raise                         # 3xx loop / 4xx: retrying will not help
            wait, err = retry_after_seconds(e.headers.get("Retry-After")) or delay, e
        except (urllib.error.URLError, http.client.HTTPException, TimeoutError, ConnectionError) as e:
            wait, err = delay, e
        if attempt == attempts:
            raise RuntimeError(f"GET failed after {attempts} attempts: {url}") from err
        if wait > MAX_WAIT:
            raise RuntimeError(f"server asked to wait {wait:.0f}s (> {MAX_WAIT}s); stopping: {url}") from err
        print(f"  retry {attempt}/{attempts - 1} in {wait:.0f}s after {err!r}")
        time.sleep(wait)
        delay *= 2
 
 
def parse_init_data(html):
    """Return the `_init_data_` object embedded in a search page, or None."""
    m = re.search(r"_init_data_\s*=\s*\{\s*data\s*:\s*", html)
    if not m:
        return None
    data, _ = json.JSONDecoder().raw_decode(html, m.end())
    return data
 
 
def obj(x):
    return x if isinstance(x, dict) else {}
 
def num(x):
    return x if isinstance(x, (int, float)) and not isinstance(x, bool) else None
 
def cents(v):
    """'109' -> 1.09; None / '' / non-digits -> None. utLogMap amounts are integer cents as strings."""
    return int(v) / 100 if isinstance(v, (int, str)) and str(v).isdigit() else None
 
 
def normalize(data):
    """-> (items, page_info). Works on the object from plain HTTP or the browser."""
    data = obj(data)
    data = obj(data.get("data", data))     # {"data": {"root": ...}} -> {"root": ...}
    fields = obj(obj(data.get("root")).get("fields"))
    page_info = obj(fields.get("pageInfo"))
    content = obj(obj(fields.get("mods")).get("itemList")).get("content")
    cards = [c for c in content if isinstance(c, dict)] if isinstance(content, list) else []
    # Display currency: from any priced card, else the page's own tracking map.
    page_currency = obj(obj(page_info.get("trace")).get("utLogMap")).get("currency")
    for c in cards:
        cur = obj(obj(c.get("prices")).get("salePrice")).get("currencyCode")
        if isinstance(cur, str):
            page_currency = cur
            break
    items = []
    for c in cards:
        pid = c.get("productId") if isinstance(c.get("productId"), str) and c.get("productId") else None
        prices = obj(c.get("prices"))
        sale, orig = obj(prices.get("salePrice")), obj(prices.get("originalPrice"))
        trace = obj(c.get("trace"))
        ut = obj(trace.get("utLogMap"))
        if num(sale.get("minPrice")) is not None:
            price, price_text, original = num(sale.get("minPrice")), sale.get("formattedPrice"), num(orig.get("minPrice"))
            currency, source = sale.get("currencyCode"), "card"
        else:                              # welcome-deal card: prices is null, amounts only in tracking
            price, price_text, original = cents(ut.get("salePriceAmount")), None, cents(ut.get("originPriceAmount"))
            currency, source = (page_currency, "tracking") if price is not None else (None, None)
        store = None
        ext = obj(trace.get("custom")).get("p4pExtendParam")
        if isinstance(ext, str):
            try:
                store = obj(json.loads(ext)).get("store_name")
            except ValueError:
                pass
        img = obj(c.get("image")).get("imgUrl")
        img = ("https:" + img if img.startswith("//") else img) if isinstance(img, str) and img else None
        items.append({
            "product_id": pid,
            "title": obj(c.get("title")).get("displayTitle"),
            "price": price,
            "price_text": price_text,
            "original_price": original,
            "currency": currency,
            "price_source": source,            # "card", "tracking" (welcome deal) or None
            "welcome_deal": "welcomegift" in str(c.get("productDetailUrl") or ""),
            "sold": obj(c.get("trade")).get("tradeDesc"),
            "sold_count": int(ut["real_trade_count"]) if str(ut.get("real_trade_count", "")).isdigit() else None,
            "rating": num(obj(c.get("evaluation")).get("starRating")),
            "store_name": store,               # only on some cards, see Gotchas
            "sponsored": "p4p" in c,
            "image": img,
            "url": f"https://www.aliexpress.com/item/{pid}.html" if pid else None,
        })
    return items, page_info
 
 
def search(query, sort=None, max_pages=3, pause=3):
    """-> (rows, report). Stops on an empty `zero_result` page or at max_pages; raises on a block."""
    max_pages = min(max_pages, SITE_MAX_PAGES)
    rows, report = {}, {"pages_read": 0, "stopped": None, "totalResults": None}
    for page in range(1, max_pages + 1):
        html = fetch(search_url(query, page, sort))
        data = parse_init_data(html)
        if data is None:
            raise SearchBlocked(f"page {page}: no _init_data_ (len={len(html)}); captcha or layout change")
        items, info = normalize(data)
        if not items:
            if info.get("searchResultType") == "zero_result":
                report["stopped"] = "zero_result"     # past the last page: the only real end
                break
            raise SearchBlocked(f"page {page}: 0 items, searchResultType={info.get('searchResultType')!r}")
        if info.get("page") not in (None, page):
            raise SearchBlocked(f"asked for page {page}, page reports {info.get('page')}")
        new = 0
        for i, it in enumerate(items):
            key = it["product_id"] or f"no-id:p{page}:{i}"   # keep id-less cards instead of dropping them
            if key not in rows:
                rows[key] = it
                new += 1
        report.update(pages_read=page, totalResults=info.get("totalResults"))
        print(f"page {page}: {len(items)} items, {new} new, totalResults={info.get('totalResults')}")
        time.sleep(pause)
    else:
        report["stopped"] = "site_cap_60" if max_pages == SITE_MAX_PAGES else f"max_pages={max_pages}"
        print(f"stopped at the page cap ({report['stopped']}), not at the end of results")
    return rows, report
 
 
rows, report = search("phone case", max_pages=3)
vals = list(rows.values())
print(len(vals), "unique products", report)
print("no product_id:", sum(r["product_id"] is None for r in vals))
print("price_source:", {s: sum(r["price_source"] == s for r in vals) for s in ("card", "tracking", None)})
print("present:", {k: sum(r[k] is not None for r in vals) for k in ("price", "original_price", "sold", "sold_count", "rating", "store_name", "image")},
      "sponsored:", sum(r["sponsored"] for r in vals))
for src in ("card", "tracking"):
    ex = next((r for r in vals if r["price_source"] == src), None)
    if ex:
        print(json.dumps(ex))

import json, re, time, urllib.parse
 
def search_url(query, page=1, sort=None):
    slug = re.sub(r"[^a-z0-9]+", "-", query.lower()).strip("-")
    params = {"SearchText": query, "page": page, **({"sortType": sort} if sort else {})}
    return f"https://www.aliexpress.com/w/wholesale-{slug}.html?{urllib.parse.urlencode(params)}"
 
# Runs in the page: reads the same _init_data_ object the server-rendered HTML embeds.
EXTRACT = r"""
(() => {
  const base = {href: location.href, ready: document.readyState};
  const d = window._dida_config_ && window._dida_config_._init_data_;
  const root = d && d.data && (d.data.data || d.data).root;
  if (!root || !root.fields) return JSON.stringify({...base, error: 'no _init_data_', title: document.title});
  const o = x => (x && typeof x === 'object' && !Array.isArray(x)) ? x : {};
  const n = x => typeof x === 'number' ? x : null;
  const f = root.fields, info = o(f.pageInfo);
  const content = o(o(f.mods).itemList).content;
  const cards = (Array.isArray(content) ? content : []).filter(c => c && typeof c === 'object');
  const sp = c => o(o(c.prices).salePrice);
  const pageCur = (cards.map(c => sp(c).currencyCode).find(x => typeof x === 'string'))
                  || o(o(info.trace).utLogMap).currency || null;
  const cents = v => (v != null && /^\d+$/.test(String(v))) ? Number(v) / 100 : null;
  return JSON.stringify({...base, info: {totalResults: info.totalResults, page: info.page, pageSize: info.pageSize,
                                         resultType: info.searchResultType},
    items: cards.map(c => {
      const pid = (typeof c.productId === 'string' && c.productId) ? c.productId : null;
      const tr = o(c.trace), ut = o(tr.utLogMap);
      let store = null;
      try { store = JSON.parse(o(tr.custom).p4pExtendParam).store_name || null } catch (e) {}
      const s = sp(c), orig = o(o(c.prices).originalPrice);
      let price, price_text = null, original, currency = null, source = null;
      if (n(s.minPrice) != null) {
        price = s.minPrice; price_text = s.formattedPrice ?? null; original = n(orig.minPrice); currency = s.currencyCode; source = 'card';
      } else {   // welcome-deal card: prices is null, amounts only in tracking
        price = cents(ut.salePriceAmount); original = cents(ut.originPriceAmount);
        if (price != null) { currency = pageCur; source = 'tracking'; }
      }
      const img = o(c.image).imgUrl;
      return {product_id: pid, title: o(c.title).displayTitle ?? null,
        price, price_text, original_price: original, currency, price_source: source,
        welcome_deal: String(c.productDetailUrl || '').includes('welcomegift'),
        sold: o(c.trade).tradeDesc ?? null, sold_count: cents(ut.real_trade_count) == null ? null : Number(ut.real_trade_count),
        rating: n(o(c.evaluation).starRating), store_name: store, sponsored: 'p4p' in c,
        image: (typeof img === 'string' && img) ? (img.startsWith('//') ? 'https:' + img : img) : null,
        url: pid ? 'https://www.aliexpress.com/item/' + pid + '.html' : null};
    })});
})()
"""
 
def open_search(url, timeout=30, attempts=2):
    """Navigate and return EXTRACT's result for THIS url's document (never the previous page's)."""
    want = urllib.parse.parse_qs(urllib.parse.urlsplit(url).query)
    for attempt in range(1, attempts + 1):
        try:
            goto_url(url)
        except TimeoutError:              # daemon IPC timeout; the navigation may still be running
            pass
        deadline = time.time() + timeout
        while time.time() < deadline:
            time.sleep(0.5)
            try:
                res = json.loads(js(EXTRACT))
            except (TimeoutError, RuntimeError):   # evaluate timed out, or the context was replaced mid-navigation
                continue
            got = urllib.parse.parse_qs(urllib.parse.urlsplit(res["href"]).query)
            if res["ready"] == "complete" and all(got.get(k) == v for k, v in want.items()):
                return res
        print(f"  attempt {attempt}: {url} did not finish loading in {timeout}s")
    raise TimeoutError(f"search page did not load: {url}")
 
def search_in_browser(query, sort=None, max_pages=3, pause=3):
    max_pages = min(max_pages, 60)        # site cap
    rows, report = {}, {"pages_read": 0, "stopped": None}
    for page in range(1, max_pages + 1):
        res = open_search(search_url(query, page, sort))
        if "error" in res:
            raise RuntimeError(f"page {page}: {res}")   # captcha or layout change: stop, do not retry in a loop
        if not res["items"]:
            if res["info"]["resultType"] == "zero_result":
                report["stopped"] = "zero_result"
                break
            raise RuntimeError(f"page {page}: 0 items, info={res['info']}")
        if res["info"]["page"] not in (None, page):
            raise RuntimeError(f"asked for page {page}, page reports {res['info']['page']}")
        new = 0
        for i, it in enumerate(res["items"]):
            key = it["product_id"] or f"no-id:p{page}:{i}"
            if key not in rows:
                rows[key] = it
                new += 1
        report["pages_read"] = page
        print(f"page {page}: {len(res['items'])} items, {new} new, info={res['info']}")
        time.sleep(pause)
    else:
        report["stopped"] = f"max_pages={max_pages}"
        print(f"stopped at the page cap ({report['stopped']}), not at the end of results")
    return rows, report
 
new_tab()                                 # blank background tab; open_search navigates it
rows, report = search_in_browser("phone case", max_pages=3)
vals = list(rows.values())
print(len(vals), "unique products", report)
print("price_source:", {s: sum(r["price_source"] == s for r in vals) for s in ("card", "tracking", None)})
ex = next((r for r in vals if r["price_source"] == "tracking"), None) or next(iter(vals), None)
print(json.dumps(ex))

import json, re, time, urllib.parse
 
class ProductAPIRefused(RuntimeError):
    """The product API answered with a captcha ("punish") instead of data. Stop and wait."""
 
def obj(x):
    return x if isinstance(x, dict) else {}
 
def capture_pdp(product_id, timeout=25):
    """Open a product page; return its `mtop.aliexpress.pdp.pc.query` result dict, or None if none arrived."""
    try:
        drain_events()                    # discard events from earlier pages
    except TimeoutError:
        pass
    try:
        goto_url(f"https://www.aliexpress.com/item/{product_id}.html")
    except TimeoutError:                  # IPC timeout; the navigation may still be running
        pass
    pending, deadline = [], time.time() + timeout
    while time.time() < deadline:
        try:
            events = drain_events()       # the daemon keeps only the last 500 events, so poll often
        except TimeoutError:
            events = []
        for e in events:
            if e.get("method") == "Network.responseReceived" and \
               "mtop.aliexpress.pdp.pc.query" in obj(obj(e.get("params")).get("response")).get("url", ""):
                pending.append(e["params"]["requestId"])
        while pending:
            rid = pending.pop(0)
            body = None
            for _ in range(20):           # body is readable once the response has finished loading
                try:
                    body = cdp("Network.getResponseBody", requestId=rid)["body"]
                    break
                except (RuntimeError, TimeoutError):
                    time.sleep(0.5)
            if body is None:
                continue
            m = re.match(r"^\s*[\w.]+\((.*)\)\s*;?\s*$", body, re.S)   # JSONP: mtopjsonp1({...})
            try:
                d = json.loads(m.group(1) if m else body)
            except ValueError:
                continue
            ret = [x for x in obj(d).get("ret") or [] if isinstance(x, str)]
            if any(x.startswith("SUCCESS") for x in ret):
                result = obj(obj(d.get("data")).get("result"))
                got = obj(result.get("PRICE")).get("productId")
                if str(got) != str(product_id):
                    print(f"  ignoring a response for product {got}, wanted {product_id}")
                    continue
                return result
            if any("TOKEN" in x for x in ret):
                continue                  # FAIL_SYS_TOKEN_EMPTY: the page sets a token cookie and retries itself
            # FAIL_SYS_USER_VALIDATE / RGV587_ERROR = slider captcha ("punish") on the API. Back off.
            raise ProductAPIRefused(f"product API refused: {ret}")
        time.sleep(0.25)
    return None
 
def parse_pdp(r):
    sku = obj(r.get("SKU"))
    prices = obj(obj(r.get("PRICE")).get("skuIdStrPriceInfoMap"))
    props_list = [p for p in sku.get("skuProperties") or [] if isinstance(p, dict)]
    names = {f'{prop.get("skuPropertyId")}:{v.get("propertyValueIdLong")}': v.get("propertyValueDisplayName")
             for prop in props_list for v in prop.get("skuPropertyValues") or [] if isinstance(v, dict)}
    variants = []
    for p in sku.get("skuPaths") or []:
        p = obj(p)
        sid = p.get("skuIdStr")
        pi = obj(prices.get(sid))
        variants.append({
            "sku_id": sid,
            "options": [names.get(k, k) for k in str(p.get("path") or "").split(";") if k],
            "price": pi.get("salePriceString"),
            "original_price": obj(pi.get("originalPrice")).get("formatedAmount"),
            "stock": p.get("skuStock"),
            "salable": p.get("salable"),
        })
    options = {prop.get("skuPropertyName"): [obj(v).get("propertyValueDisplayName") for v in prop.get("skuPropertyValues") or []]
               for prop in props_list}
    span = lambda a, b, sep: f"{a}{sep}{b}" if a is not None and b is not None else None
    shipping = []
    for o in obj(r.get("SHIPPING")).get("deliveryLayoutInfo") or []:
        b = obj(obj(o).get("bizData"))
        shipping.append({
            "method": b.get("deliveryProviderName") or b.get("company"),
            "fee": "Free" if b.get("shippingFee") == "free" else b.get("formattedAmount"),
            "ship_from": b.get("shipFrom"),
            "ship_to": b.get("shipToCode"),
            "eta": span(b.get("displayEtaMinDate"), b.get("displayEtaMaxDate"), " - "),
            "days": span(b.get("deliveryDayMin"), b.get("deliveryDayMax"), "-"),
            "tracking": b.get("tracking"),
        })
    shop = obj(r.get("SHOP_CARD_PC"))
    seller = obj(shop.get("sellerInfo"))
    rating = obj(r.get("PC_RATING"))
    props = obj(r.get("PRODUCT_PROP_PC"))
    store_url = seller.get("storeURL") if isinstance(seller.get("storeURL"), str) else ""
    pid = obj(r.get("PRICE")).get("productId")
    return {
        "product_id": str(pid) if pid is not None else None,
        "title": obj(r.get("PRODUCT_TITLE")).get("text"),
        "rating": rating.get("rating"),
        "reviews": rating.get("totalValidNum"),
        "sales_text": rating.get("otherText"),
        "store_name": shop.get("storeName"),
        "store_url": ("https:" + store_url if store_url.startswith("//") else store_url) or None,
        "store_positive_rate": shop.get("sellerPositiveRate"),
        "store_open_since": seller.get("formatOpenTime"),
        "options": options,
        "variants": variants,
        "shipping": shipping,
        "images": [i for i in obj(r.get("HEADER_IMAGE_PC")).get("imagePathList") or [] if isinstance(i, str)],
        "specs": [(obj(a).get("attrName"), obj(a).get("attrValue")) for a in props.get("showedProps") or props.get("outerProps") or []],
    }
 
def search_ids(url, timeout=30):
    """Open a search page and return its product IDs, after confirming the tab shows that URL's query."""
    want = urllib.parse.parse_qs(urllib.parse.urlsplit(url).query)
    goto_url(url)
    deadline = time.time() + timeout
    while time.time() < deadline:
        time.sleep(0.5)
        try:
            res = json.loads(js("""JSON.stringify((() => { const d = window._dida_config_ && window._dida_config_._init_data_;
              const root = d && d.data && (d.data.data || d.data).root;
              const c = root && root.fields && root.fields.mods && root.fields.mods.itemList && root.fields.mods.itemList.content;
              return {href: location.href, ready: document.readyState,
                      ids: Array.isArray(c) ? c.map(x => x && x.productId).filter(x => typeof x === 'string') : null}; })())"""))
        except (TimeoutError, RuntimeError):
            continue
        got = urllib.parse.parse_qs(urllib.parse.urlsplit(res["href"]).query)
        if res["ready"] == "complete" and all(got.get(k) == v for k, v in want.items()):
            if res["ids"] is None:
                raise RuntimeError(f"search page has no _init_data_: {res['href']}")
            return res["ids"]
    raise TimeoutError(f"search page did not load: {url}")
 
new_tab()
ids = search_ids("https://www.aliexpress.com/w/wholesale-mechanical-keyboard.html?SearchText=mechanical+keyboard")
print(len(ids), "ids from search")
if not ids:
    raise RuntimeError("search returned no product IDs")
parsed, no_response, refused_at = [], [], None
for pid in ids[:3]:
    time.sleep(8)                         # polite pacing between product pages
    try:
        r = capture_pdp(pid)
    except ProductAPIRefused as e:
        refused_at = pid
        print(pid, e, "- stopping; wait several minutes before more product pages")
        break
    if r is None:
        no_response.append(pid)
        print(pid, "no product API response seen (layout change?)", page_info()["url"])
        continue
    d = parse_pdp(r)
    parsed.append(d)
    print(json.dumps({**d, "images": d["images"][:2], "variants": d["variants"][:3], "specs": d["specs"][:3],
                      "n_variants": len(d["variants"]), "n_images": len(d["images"]), "n_specs": len(d["specs"])}))
print(f"parsed {len(parsed)} of {len(ids[:3])}; no response: {no_response}; refused at: {refused_at}")

{"product_id": "3256807353643961",
 "title": "AJAZZ AK820 and AK820PRO Mechanical Gaming Keyboard 75% Layout PBT OEM Keycap RGB Backlight Full Key Anti-Ghosting Type-C",
 "rating": "4.9", "reviews": 5048, "sales_text": "This seller: 700+ sales | Total sales: 10,000+",
 "store_name": "Stone's Store", "store_url": "https://www.aliexpress.com/store/1103576287",
 "store_positive_rate": "100.0", "store_open_since": "Mar 1, 2024",
 "options": {"Color": ["AK820 Gray", "AK820Pro Gift Purple", "AK820Pro Gift Gray", "AK820 Purple", "AK820 Black"], "Axis Body": ["Red Switch"]},
 "variants": [{"sku_id": "12000061215413600", "options": ["AK820Pro Gift Purple", "Red Switch"], "price": "$77.93", "original_price": "$170.69", "stock": 10, "salable": true},
              {"sku_id": "12000041213380330", "options": ["AK820 Black", "Red Switch"], "price": "$25.32", "original_price": "$61.08", "stock": 68, "salable": true}],
 "shipping": [{"method": "AliExpress Selection Standard Shipping", "fee": "$1.99", "ship_from": "China", "ship_to": "US",
               "eta": "Oct. 29 - Nov. 03", "days": "6-11", "tracking": "visible"}],
 "n_variants": 5, "n_images": 6, "n_specs": 15}