# -*- coding: utf-8 -*- """扩展测试 v2：修复邮编持久化/变体/BSR/Coupon/评论总数(滚动加载)。UK+DE。""" 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_alt": "https://www.amazon.co.uk/product-reviews/B08GY1VLDK/?filterByStar=five_star&reviewerType=all_reviews", "postcode": "SW1A 1AA", "locale": "en-GB", "tz": "Europe/London", "domain": ".amazon.co.uk", "currency": "GBP", }, "DE": { "url": "https://www.amazon.de/dp/B08GY1VLDK", "reviews_alt": "https://www.amazon.de/product-reviews/B08GY1VLDK/?filterByStar=five_star&reviewerType=all_reviews", "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, timeout=2500): try: el = page.locator(sel).first return el.inner_text(timeout=timeout).strip() if el.count() else None except Exception: return None def glow_text(page): return " ".join((t(page, "#nav-global-location-popover-link") or "").split()) def set_zip_verified(page, zip_): """设邮编并验证 glow 已更新，最多试 2 次""" for attempt in range(2): 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(7) for csel in ["#GLUXConfirmClose", "input[name='glowDone']"]: try: el = page.locator(csel).first if el.count() and el.is_visible(): el.click(timeout=2000); time.sleep(2) except Exception: pass page.reload(wait_until="domcontentloaded", timeout=45000) time.sleep(3) g = glow_text(page) digits = re.sub(r"\D", "", zip_) if digits and digits in re.sub(r"\D", "", g): return True, g except Exception as e: pass return False, glow_text(page) def extract(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["BuyBox销售店铺"] = t(page, "#sellerProfileTriggerId") d["可售状态"] = " ".join((t(page, "#availability") or "").split()) # --- Coupon：可见 DOM 文本，限定价格/促销区域 --- coupon_texts = [] for sel in ["#couponBadgeRegularVpc", "#vpsButton", ".coupon-badge", "#promoPriceBlockMessage", "#corePriceDisplay_div .promo-message", "span.promo-message"]: v = t(page, sel) if v: coupon_texts.append(" ".join(v.split())[:80]) # 全页面可见文本兜底搜关键词（限制在 body 文本，不含 style/script） try: body_txt = page.evaluate("() => document.body.innerText") for kw in ["Coupon", "Limited time deal", "Rabatt", "Angebot"]: for m in re.finditer(re.escape(kw), body_txt): seg = " ".join(body_txt[max(0, m.start()-20):m.start()+90].split()) coupon_texts.append(seg) if len(coupon_texts) > 8: break except Exception: pass d["Coupon可见文本"] = list(dict.fromkeys(x for x in coupon_texts if x))[:8] or None # --- 变体：asin 在 li 上 --- variants = [] try: lis = page.locator("#twister li[data-asin], #twister li[data-defaultasin], #twister ul li img") n = lis.count() for i in range(min(n, 25)): li = lis.nth(i) a = li.get_attribute("data-asin") or li.get_attribute("data-defaultasin") if not a: img = li.locator("img").first if li.locator("img").count() else li alt = (img.get_attribute("alt") if img else None) or li.inner_text(timeout=1000) a = f"(无asin:{' '.join((alt or '').split())[:30]})" cls = (li.get_attribute("class") or "") state = "选中" if "selected" in cls else ("不可选" if "unavailable" in cls else "可选") variants.append(f"{a}|{state}") except Exception as e: variants = [f"ERR {type(e).__name__}"] d["变体清单"] = variants[:25] d["变体数量"] = len(variants) html = page.content() m = re.search(r'"parentAsin"\s*:\s*"([A-Z0-9]{10})"', html) d["父ASIN"] = m.group(1) if m else None # --- BSR：逐 li 找 --- bsr = [] try: lis = page.locator("#detailBulletsWrapper_feature_div li, #productDetails_detailBullets_sections1 li, #detailBullets_feature_div li") for i in range(lis.count()): txt = " ".join(lis.nth(i).inner_text().split()) if re.search(r"Best\s*Sellers?\s*Rank|Bestseller-?Rang", txt, re.I): bsr.append(txt[:200]) if not bsr: # 表格版 rows = page.locator("#productDetails_techSpec_section_1 tr, #productDetails_techSpec_section_2 tr, table.prodDetTable tr") for i in range(rows.count()): txt = " ".join(rows.nth(i).inner_text().split()) if re.search(r"Best\s*Sellers?\s*Rank|Bestseller-?Rang", txt, re.I): bsr.append(txt[:200]) except Exception: pass d["BSR"] = bsr or None # --- 类目去重 --- try: bc = page.locator("#wayfinding-breadcrumbs_feature_div li") seen, path = set(), [] for i in range(bc.count()): x = " ".join(bc.nth(i).inner_text().split()) if x and x != "›" and x not in seen: seen.add(x); path.append(x) d["类目路径"] = path[:8] except Exception: pass # --- 评论总数：滚到评论区触发懒加载 --- try: page.evaluate("() => document.querySelector('#reviewsMedley, #cm-cr-dp-review-list, #customerReviews')?.scrollIntoView()") time.sleep(2) fi = page.locator('[data-hook="cr-filter-info-review-rating-count"]') if fi.count(): d["评论总数行"] = " ".join(fi.first.inner_text().split())[:150] except Exception: pass d["评分人数"] = t(page, "#acrCustomerReviewText") h = t(page, "#histogramTable") d["星级分布"] = " ".join(h.split())[:180] if h else None d["配送信息"] = " ".join((t(page, "#deliveryMessage") or t(page, "#mir-layout-DELIVERY_BLOCK") or "").split())[:130] 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)) ok, g = set_zip_verified(page, cfg["postcode"]) r["邮编生效"] = f"{ok} | {g}" r.update(extract(page)) # 备用评论页格式试一次 try: page.goto(cfg["reviews_alt"], wait_until="domcontentloaded", timeout=45000) time.sleep(2.5) title = t(page, "title") fi = page.locator('[data-hook="cr-filter-info-review-rating-count"]') r["product-reviews格式"] = f"title={title[:60]} | filter行={' '.join(fi.first.inner_text().split())[:120] if fi.count() else '无'}" except Exception as e: r["product-reviews格式"] = f"ERR {type(e).__name__}" 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()