# -*- coding: utf-8 -*- """商品页字段抽取：产出与「抓取与对比记录表」列名对齐的清洗后字段。 原始值保留在 价格证据/原文类字段；数值字段按站点清洗器归一。 """ import json, re, time from sites import SITE_CONFIGS, CLEANERS, clean_histogram 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 _norm_img_url(u, token="_SL1500_"): """把 m.media-amazon.com 图片 URL 规格化为固定 token，便于逐轮比较""" if not u or not u.startswith("http"): return u return re.sub(r"\._[^_]+_\.jpg$", f".{token}_.jpg", u.split(" ")[0]) def parse_variants_from_state(page): """从 twister 的 a-state JSON 解析变体：子ASIN|维度=值|状态""" try: scripts = page.locator("#twister_feature_div script[data-a-state]") for i in range(scripts.count()): raw = scripts.nth(i).text_content() or "" try: data = json.loads(raw) except Exception: continue dims = data.get("sortedDimValuesForAllDims") or data.get("dimensionValuesDisplayData") if not dims: continue items = [] for dim, vals in dims.items(): for v in vals: if not isinstance(v, dict): continue asin = v.get("defaultAsin") or v.get("asin") state = v.get("dimensionValueState") or "" disp = v.get("dimensionValueDisplayText") or v.get("displayValue") or "" items.append({"维度": dim, "子ASIN": asin, "属性": disp, "状态": state}) if items: return items except Exception: pass return None def collect_promo_lines(page): """购买列文本，截断到'经常一起买/相关商品'分界，保留促销关键词行。""" txt = _t(page, "#centerCol") or "" for stop in ["Frequently bought together", "Häufig zusammen gekauft", "Products related to this item", "Ähnliche Artikel", "Compare with similar items", "Vergleichen"]: idx = txt.find(stop) if idx > 0: txt = txt[:idx] kw = re.compile(r"(Limited time deal|Angebot des Tages|Befristetes Angebot|Deal of the Day|Lightning Deal|Coupon|Save\s*[£€]|Spare\w*\s*[£€]|percent savings|%?\s*Rabatt|\d+%\s*off)", re.I) keep = [] for line in txt.split("\n"): line = " ".join(line.split()) if line and kw.search(line): keep.append(line[:120]) return " | ".join(keep)[:400] def extract_product(page, site): cfg = SITE_CONFIGS[site] cl = CLEANERS[site] d = {} d["实际站点"] = site d["实际币种"] = cfg["currency"] # ---- 标题/链接/ASIN ---- d["标题"] = " ".join((_t(page, "#productTitle") or "").split()) or None d["实际页面链接"] = page.url m = re.search(r"/dp/([A-Z0-9]{10})", page.url) d["实际ASIN"] = m.group(1) if m else None html = page.content() m = re.search(r'"parentAsin"\s*:\s*"([A-Z0-9]{10})"', html) d["父ASIN"] = m.group(1) if m else None # ---- 价格组（原始+清洗） ---- raw_price = _t(page, "#corePriceDisplay_div .a-price .a-offscreen") or _t(page, "span.a-price .a-offscreen") raw_strike = _t(page, "#corePriceDisplay_div .priceBlockStrikePriceString") or _t(page, "span.a-price.a-text-price .a-offscreen") raw_unit = _t(page, ".basisPrice .a-offscreen") d["当前展示价格"] = cl["price"](raw_price) d["历史参考价"] = cl["price"](raw_strike) # Was/UVP 划线价 d["单位价格"] = cl["price"](raw_unit) m = re.search(r"(?:UVP|RRP)[:\s]*[£€]?\s*([\d.,]+)", html) d["建议零售价"] = cl["price"](m.group(1)) if m else None d["价格证据"] = "; ".join(x for x in [f"展示:{raw_price}", f"划线:{raw_strike}", f"单位:{raw_unit}"] if x and x.split(":", 1)[1] != "None") # ---- BuyBox 组 ---- d["当前销售店铺"] = _t(page, "#sellerProfileTriggerId") d["Buy Box状态"] = ("有featured offer(" + (d["当前销售店铺"] or "?") + ")") if d["当前销售店铺"] else "无featured offer" btns = [] for sel, name in [("#add-to-cart-button", "Add to basket"), ("#buy-now-button", "Buy Now")]: try: b = page.locator(sel) if b.count(): btns.append(f"{name}:{'可见启用' if b.first.is_visible() and not b.first.is_disabled() else '隐藏或禁用'}") else: btns.append(f"{name}:未找到") except Exception: btns.append(f"{name}:未检查") d["购买按钮状态"] = "; ".join(btns) d["可售状态"] = " ".join((_t(page, "#availability") or "").split()) or None d["配送信息原文"] = " ".join((_t(page, "#deliveryMessage") or _t(page, "#mir-layout-DELIVERY_BLOCK") or "").split())[:200] or None # ---- 促销/Coupon（购买列、截断防广告污染） ---- blob = collect_promo_lines(page) d["Coupon及促销原文"] = blob or None lt = re.search(r"(Limited time deal|Angebot des Tages|Befristetes Angebot|Deal of the Day|Lightning Deal)", blob) d["限时优惠标识"] = lt.group(1) if lt else None m = re.search(r"(?:Save|Sparen|Spare)\s*[£€]?\s*([\d.,]+)", blob) d["优惠券金额"] = cl["price"](m.group(1)) if m else None m = re.search(r"(\d+)\s*(?:percent savings|% off|% Rabatt|Prozent)", blob, re.I) d["优惠券折扣百分比"] = cl["number"](m.group(1)) if m else None # ---- 评论/评分 ---- try: rt = page.locator("#acrPopover").first.get_attribute("title", timeout=1500) if page.locator("#acrPopover").count() else None except Exception: rt = None d["综合评分"] = cl["rating"](rt or _t(page, "span[data-hook='rating-out-of-text']")) d["评分人数"] = cl["count"](_t(page, "#acrCustomerReviewText")) hist_raw = _t(page, "#histogramTable") d["_星级百分比"] = clean_histogram(site, hist_raw) # 诊断用；表内1~5星数留空(登录墙) try: page.evaluate("() => document.querySelector('#reviewsMedley, #cm-cr-dp-review-list, #customerReviews')?.scrollIntoView()") time.sleep(1.5) fi = _t(page, '[data-hook="cr-filter-info-review-rating-count"]') if fi: d["_评论过滤行原文"] = " ".join(fi.split())[:150] mm = re.search(r"([\d.,]+)\s+(?:with reviews|mit Rezensionen)", fi) or re.search(r"^([\d.,]+)", fi) d["评论总数"] = cl["count"](mm.group(1)) if mm else None except Exception: pass # ---- 图片/视频 ---- try: main = page.locator("#landingImage").first d["主图URL"] = _norm_img_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, vids = [], 0 for i in range(imgs.count()): u = imgs.nth(i).get_attribute("src") if not u or "sprite" in u or "transparent-pixel" in u: continue is_vid = False try: li = imgs.nth(i).locator("xpath=ancestor::li[1]") li_html = li.inner_html(timeout=800) if li.count() else "" is_vid = bool(re.search(r"video-block-injected|a-button-toggle-video|videoCount|video-count", li_html)) except Exception: pass if is_vid: vids += 1 else: urls.append(_norm_img_url(u)) d["副图URL列表"] = urls d["副图数量"] = len(urls) d["视频展示数量"] = vids if vids else None except Exception: pass try: apl = page.locator("#aplus_feature_div img, #aplus img") aus = [_norm_img_url(apl.nth(i).get_attribute("src")) for i in range(apl.count())] aus = [u for u in aus if u and u.startswith("http") and "sprite" not in u and "grey-pixel" not in u] d["A+图片URL列表"] = aus except Exception: pass # ---- 类目/BSR ---- 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["所属类目路径"] = " > ".join(path[:8]) or None except Exception: pass ranks = [] 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): ranks = re.findall(r"(?:#|Nr\.?)\s*([\d.,]+)\s+in\s+([^\n(;]{2,60})", txt) d["_BSR原文"] = txt[:250] break except Exception: pass if ranks: d["大类排名"] = f"#{ranks[0][0]} in {ranks[0][1].strip()}" d["小类排名"] = "; ".join(f"#{n} in {c.strip()}" for n, c in ranks[1:]) or None # ---- 卖点 ---- 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())][:12] except Exception: pass # ---- 变体（a-state JSON 优先，DOM 兜底） ---- vs = parse_variants_from_state(page) if not vs: vs = [] try: lis = page.locator("li[data-asin], li[data-defaultasin]") for i in range(lis.count()): a = lis.nth(i).get_attribute("data-asin") or lis.nth(i).get_attribute("data-defaultasin") cls = lis.nth(i).get_attribute("class") or "" vs.append({"子ASIN": a, "状态": "SELECTED" if "selected" in cls else ("UNAVAILABLE" if "unavailable" in cls else "AVAILABLE")}) except Exception: pass d["变体清单"] = [f"{v['子ASIN']}|{v.get('维度','')}={v.get('属性','')}|{v.get('状态','')}" for v in (vs or [])] d["变体数量"] = len(d["变体清单"]) # ---- 访问状态 ---- d["页面访问状态"] = "正常商品页" if d.get("标题") else "错误页面" return d