# -*- coding: utf-8 -*- """扩展测试 v1：验证大字段组抓取能力（价格组/BuyBox/变体/图片A+视频/类目BSR/卖点/评论分布/评论页filterByStar） UK + DE 各跑 A32 商品。输出 /home/user/extended_result.json """ import json, re, time, random from playwright.sync_api import sync_playwright SITES = { "UK": { "url": "https://www.amazon.co.uk/Auckly-Wireless-Electromagnetic-Automatic-Compatible/dp/B08GY1VLDK", "reviews": "https://www.amazon.co.uk/portal/customer-reviews/B08GY1VLDK", "postcode": "SW1A 1AA", "locale": "en-GB", "tz": "Europe/London", "domain": ".amazon.co.uk", "currency": "GBP", }, "DE": { "url": "https://www.amazon.de/dp/B08GY1VLDK", "reviews": "https://www.amazon.de/-/en/portal/customer-reviews/B08GY1VLDK", "postcode": "10115", "locale": "de-DE", "tz": "Europe/Berlin", "domain": ".amazon.de", "currency": "EUR", }, } UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36" def t(page, sel): try: el = page.locator(sel).first return el.inner_text(timeout=2500).strip() if el.count() else None except Exception: return None def set_zip(page, zip_): try: page.click("#nav-global-location-popover-link", timeout=8000) page.wait_for_selector("#GLUXZipUpdateInput", state="visible", timeout=8000) page.fill("#GLUXZipUpdateInput", zip_) page.click("#GLUXZipUpdate", timeout=4000, force=True) time.sleep(4) return True except Exception: return False def parse_price(s): if not s: return None s2 = s.replace("£", "").replace("€", "").replace(",", ".").strip() try: return round(float(re.sub(r"[^\d.]", "", s2)), 2) except Exception: return None def extract_product_page(page): d = {} # --- 价格组 --- d["当前展示价格"] = t(page, "#corePriceDisplay_div .a-price .a-offscreen") or t(page, "span.a-price .a-offscreen") d["促销价格"] = t(page, "#corePriceDisplay_div .priceBlockStrikePriceString") or t(page, "span.a-price.a-text-price .a-offscreen") d["单价"] = t(page, "#corePriceDisplay_div .basisPrice .a-offscreen") or t(page, ".basisPrice .a-offscreen") d["价格证据"] = t(page, "#corePriceDisplay_div") # --- BuyBox 组 --- d["BuyBox销售店铺"] = t(page, "#sellerProfileTriggerId") d["购买按钮_加购"] = None try: b = page.locator("#add-to-cart-button") if b.count(): d["购买按钮_加购"] = f"visible={b.first.is_visible()},disabled={b.first.is_disabled()}" except Exception: pass d["购买按钮_立即买"] = None try: b = page.locator("#buy-now-button") if b.count(): d["购买按钮_立即买"] = f"visible={b.first.is_visible()},disabled={b.first.is_disabled()}" except Exception: pass d["可售状态"] = " ".join((t(page, "#availability") or "").split()) d["配送信息"] = " ".join((t(page, "#deliveryMessage") or t(page, "#mir-layout-DELIVERY_BLOCK") or "").split())[:120] # --- Coupon / 促销 --- html = page.content() coupon = re.findall(r"(Coupon[^<]{0,80}|Save\s+[£€][\d.,]+[^<]{0,40}|Spare?n?\s+[£€][\d.,]+[^<]{0,40}|\d+\s*%\s*(?:off|rabatt|Rabatt)[^<]{0,40}|Limited time deal|Angebot des Tages)", html) d["Coupon促销片段"] = list(dict.fromkeys(x.strip() for x in coupon))[:6] or None # --- 变体 --- try: swatch_imgs = page.locator("#twister img[data-asin]") asins = [] for i in range(swatch_imgs.count()): a = swatch_imgs.nth(i).get_attribute("data-asin") alt = swatch_imgs.nth(i).get_attribute("alt") cls = swatch_imgs.nth(i).get_attribute("class") or "" state = "选中" if "selected" in (swatch_imgs.nth(i).get_attribute("class") or "") else ("不可选" if "swatch-unavailable" in cls else "可选") asins.append({"asin": a, "alt": alt, "state": state}) d["变体清单(img)"] = asins[:20] d["变体数量"] = len(asins) except Exception as e: d["变体清单(img)"] = f"ERR {type(e).__name__}" m = re.search(r'"parentAsin"\s*:\s*"([A-Z0-9]{10})"', html) d["父ASIN"] = m.group(1) if m else None # --- 图片/视频/A+ --- try: main = page.locator("#landingImage").first d["主图URL"] = main.get_attribute("data-old-hires") or main.get_attribute("src") except Exception: d["主图URL"] = None try: imgs = page.locator("#altImages li.item img") urls = [imgs.nth(i).get_attribute("src") for i in range(imgs.count())] urls = [u for u in urls if u and "sprite" not in u and "transparent-pixel" not in u] d["副图URL列表"] = urls[:15] d["副图数量"] = len(urls) except Exception: pass try: vids = page.locator("#altImages li:has(.a-button-toggle-video), #altImages li[data-csa-c-content-id*='video']") d["视频角标数"] = vids.count() except Exception: pass try: apl = page.locator("#aplus_feature_div img, #aplus img") aus = [apl.nth(i).get_attribute("src") for i in range(apl.count())] aus = [u for u in aus if u and "sprite" not in u and "grey-pixel" not in u and u.startswith("http")] d["A+图片URL列表"] = aus[:20] d["A+图片数量"] = len(aus) except Exception: pass # --- 类目/BSR --- try: bc = page.locator("#wayfinding-breadcrumbs_feature_div li a, #wayfinding-breadcrumbs_feature_div li span.a-list-item") d["类目路径"] = [bc.nth(i).inner_text().strip() for i in range(bc.count())][:8] except Exception: pass bullets = t(page, "#detailBulletsWrapper_feature_div") or t(page, "#productDetails_detailBullets_sections1") or "" ranks = re.findall(r"#?([\d.,]+)\s+in\s+([^\n(]{2,60}?)\s*(?:\(|$)", bullets) if ranks: d["BSR列表"] = [f"#{r} in {c.strip()}" for r, c in ranks[:8]] # --- 卖点 --- try: fb = page.locator("#feature-bullets li span.a-list-item") d["卖点列表"] = [" ".join(fb.nth(i).inner_text().split()) for i in range(fb.count())][:10] except Exception: pass # --- 评论分布(商品页 histogram %) --- try: hist = page.locator("#histogramTable td.a-text-right, #histogramTable span.a-letter-space") rows = page.locator('[data-hook="cr-filter-info-review-rating-count"]') d["评论概要行"] = " ".join((rows.first.inner_text() if rows.count() else "").split())[:120] h = page.locator("#histogramTable").first d["星级分布原始"] = " ".join(h.inner_text().split())[:200] if h.count() else None except Exception: pass return d def run(): out = {} with sync_playwright() as p: browser = p.chromium.launch(headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"]) for site, cfg in SITES.items(): r = {} ctx = browser.new_context(user_agent=UA, locale=cfg["locale"], timezone_id=cfg["tz"], viewport={"width": 1366, "height": 768}) ctx.add_cookies([{"name": "i18n-prefs", "value": cfg["currency"], "domain": cfg["domain"], "path": "/"}]) page = ctx.new_page() try: page.goto(cfg["url"], wait_until="domcontentloaded", timeout=45000) time.sleep(random.uniform(2, 3)) r["设邮编"] = set_zip(page, cfg["postcode"]) page.reload(wait_until="domcontentloaded", timeout=45000) time.sleep(random.uniform(2.5, 3.5)) r["商品页"] = extract_product_page(page) r["配送地"] = " ".join((t(page, "#nav-global-location-popover-link") or "").split()) # --- 评论页测试：总数 + filterByStar=five_star --- rv = {} try: page.goto(cfg["reviews"], wait_until="domcontentloaded", timeout=45000) time.sleep(random.uniform(2, 3)) fi = page.locator('[data-hook="cr-filter-info-review-rating-count"]') rv["评论总数行"] = " ".join((fi.first.inner_text() if fi.count() else (t(page, "title") or "")).split())[:150] page.goto(cfg["reviews"] + "?filterByStar=five_star", wait_until="domcontentloaded", timeout=45000) time.sleep(random.uniform(2, 3)) fi2 = page.locator('[data-hook="cr-filter-info-review-rating-count"]') rv["五星筛选行"] = " ".join((fi2.first.inner_text() if fi2.count() else (t(page, "title") or "")).split())[:150] # 星级汇总条 hs = page.locator('[data-hook="cr-filter-info-review-rating-count"]') rv["评论页标题"] = t(page, "title") except Exception as e: rv["错误"] = f"{type(e).__name__}: {e}" r["评论页"] = rv r["状态"] = "OK" except Exception as e: r["状态"] = f"异常: {e}" finally: out[site] = r ctx.close() time.sleep(random.uniform(2, 4)) browser.close() print(json.dumps(out, ensure_ascii=False, indent=2)) if __name__ == "__main__": run()