catalog-extract/scripts/parse_product.py · py · download
#!/usr/bin/env python3
"""Generic product-page parser: HTML -> canonical product row + extra_attributes.
Order of preference (JSON is stable, DOM selectors rot):
1. JSON-LD `Product` (bare, in a list, or under `@graph`) -> name, sku, price, currency,
availability, brand, image(s),
description, rating, reviews
2. `__NEXT_DATA__` -> returned raw for a hook to map
3. Spec harvest from the DOM: <table> rows, <dl> pairs, and
label/value sibling pairs ("Case material: Stainless steel") -> extra_attributes, then
promoted via a spec map
4. Fallbacks: <title>, og:*, itemprop=price, meta[property=product:price:amount]
A spec map (JSON file) promotes site labels to canonical columns:
{"case material": "case_material", "movement": "movement_type",
"water resistance": "water_resistance", "dial colour": "dial_color"}
Labels are matched lower-cased and stripped of trailing ':'. Every unmapped
label is still kept in extra_attributes, so nothing published is dropped and
`audit` can show you what to promote next.
CLI:
python parse_product.py page.html [--spec-map specs.json] [--url URL]
python parse_product.py page.html --labels # just list spec labels found
"""
from __future__ import annotations
import argparse
import json
import re
import sys
from pathlib import Path
from bs4 import BeautifulSoup
# ----------------------------------------------------------------------------- JSON payloads
_NEXT_DATA_RE = re.compile(r'<script id="__NEXT_DATA__"[^>]*>(.*?)</script>', re.DOTALL)
def extract_json_ld(html: str) -> tuple[dict, dict]:
"""(first Product node, first BreadcrumbList node); {} when absent."""
product, breadcrumb = {}, {}
soup = BeautifulSoup(html, "html.parser")
for tag in soup.find_all("script", {"type": "application/ld+json"}):
raw = tag.string or tag.get_text() or ""
if not raw.strip():
continue
try:
data = json.loads(raw)
except (json.JSONDecodeError, TypeError):
# some sites leave a trailing comma or comment; try a lenient slice
try:
data = json.loads(raw[raw.find("{"): raw.rfind("}") + 1])
except (json.JSONDecodeError, TypeError, ValueError):
continue
for node in _iter_ld(data):
t = node.get("@type")
types = t if isinstance(t, list) else [t]
if "Product" in types and not product:
product = node
elif "BreadcrumbList" in types and not breadcrumb:
breadcrumb = node
return product, breadcrumb
def _iter_ld(data):
if isinstance(data, dict):
if isinstance(data.get("@graph"), list):
for n in data["@graph"]:
if isinstance(n, dict):
yield n
return
yield data
# nested mainEntity / itemListElement products
for v in data.values():
if isinstance(v, (dict, list)):
yield from _iter_ld(v)
elif isinstance(data, list):
for n in data:
yield from _iter_ld(n)
def extract_next_data(html: str) -> dict | None:
m = _NEXT_DATA_RE.search(html or "")
if not m:
return None
try:
return json.loads(m.group(1))
except json.JSONDecodeError:
return None
_OPEN = {"{": "}", "[": "]"}
def extract_js_json(html: str, anchor: str, *, unescape_first: bool = False):
"""First balanced {...}/[...] after regex `anchor` — for `window.__STATE__ = {...}`,
`dataLayer.push({...})`, `window[Symbol.for("InstantSearchInitialResults")] = {...}`."""
if not html:
return None
sources = [html] if not unescape_first else [html, html.replace('\\"', '"')]
for src in sources:
m = re.search(anchor, src)
if not m:
continue
blob = _balanced(src, m.end())
if blob is None:
continue
try:
return json.loads(blob)
except (json.JSONDecodeError, TypeError):
continue
return None
def _balanced(text: str, start: int) -> str | None:
i, n = start, len(text)
while i < n and text[i] not in _OPEN:
if text[i] not in " \t\r\n=:(,":
return None
i += 1
if i >= n:
return None
opener, closer = text[i], _OPEN[text[i]]
depth, j, in_str, esc = 0, i, False, False
while j < n:
c = text[j]
if in_str:
if esc:
esc = False
elif c == "\\":
esc = True
elif c == '"':
in_str = False
else:
if c == '"':
in_str = True
elif c == opener:
depth += 1
elif c == closer:
depth -= 1
if depth == 0:
return text[i:j + 1]
j += 1
return None
# ----------------------------------------------------------------------------- scalars
_CURRENCY = {"$": "USD", "US$": "USD", "USD": "USD", "€": "EUR", "EUR": "EUR", "£": "GBP",
"GBP": "GBP", "CHF": "CHF", "¥": "JPY", "JPY": "JPY", "CA$": "CAD", "CAD": "CAD",
"A$": "AUD", "AUD": "AUD"}
_NUM_RE = re.compile(r"(\d{1,3}(?:[,\s]\d{3})+|\d+)(?:[.,](\d{1,2}))?")
def parse_price(raw) -> tuple[float | None, str | None]:
"""Number -> (float, 'USD'); '$1,234.00' / 'USD 850' -> (value, currency);
0 or empty -> (None, None) (a genuine gap, never a zero price)."""
if raw is None or isinstance(raw, bool):
return None, None
if isinstance(raw, (int, float)):
return (None, None) if raw == 0 else (float(raw), "USD")
if not isinstance(raw, str) or not raw.strip():
return None, None
text = raw.strip()
currency = None
for sym, code in sorted(_CURRENCY.items(), key=lambda kv: -len(kv[0])):
if sym in text:
currency = code
break
m = _NUM_RE.search(text)
if not m:
return None, None
whole = re.sub(r"[,\s]", "", m.group(1))
frac = m.group(2) or "0"
try:
v = float(f"{whole}.{frac}")
except ValueError:
return None, None
return (None, None) if v == 0 else (v, currency)
_MM_Q_RE = re.compile(r"(?:diameter|Ø|case size|size)\s*[:：]?\s*(\d+(?:[.,]\d+)?)\s*mm", re.I)
_MM_RE = re.compile(r"(\d+(?:[.,]\d+)?)\s*(?:mm|millimet)", re.I)
_FLOAT_RE = re.compile(r"(\d+(?:[.,]\d+)?)")
def parse_mm(value) -> float | None:
if isinstance(value, bool) or value is None:
return None
if isinstance(value, (int, float)):
return round(float(value), 1)
s = str(value)
m = _MM_Q_RE.search(s) or _MM_RE.search(s) or _FLOAT_RE.search(s)
if not m:
return None
try:
return round(float(m.group(1).replace(",", ".")), 1)
except ValueError:
return None
def norm_gender(value) -> str:
v = (str(value or "")).strip().lower()
if not v:
return ""
if "unisex" in v:
return "unisex"
if v.startswith("wom") or "ladies" in v or v in ("female", "f", "lady"):
return "women's"
if v.startswith("men") or v in ("male", "m", "man", "gents", "gentlemen"):
return "men's"
return ""
def norm_availability(value) -> str:
v = str(value or "")
tail = v.rsplit("/", 1)[-1]
return tail.replace("https://schema.org/", "").strip() or ""
# ----------------------------------------------------------------------------- spec harvest
_LABEL_CLASS_RE = re.compile(r"(label|name|title|key|attr(ibute)?-?name|spec-?name|term)", re.I)
_VALUE_CLASS_RE = re.compile(r"(value|data|attr(ibute)?-?value|spec-?value|desc)", re.I)
def _txt(el) -> str:
return re.sub(r"\s+", " ", el.get_text(" ", strip=True)).strip()
def _norm_label(s: str) -> str:
return re.sub(r"\s+", " ", s).strip().rstrip(":").strip().lower()
def harvest_specs(soup: BeautifulSoup) -> dict[str, str]:
"""Every label/value pair the page prints, keyed by the label as printed
(lower-cased). Sources: tables, definition lists, class-named label/value
siblings, and 'Label: value' list items. First value per label wins."""
specs: dict[str, str] = {}
def put(label: str, value: str) -> None:
label, value = _norm_label(label), re.sub(r"\s+", " ", value).strip()
if label and value and len(label) <= 60 and len(value) <= 400 and label not in specs:
specs[label] = value
for tr in soup.select("table tr"):
cells = tr.find_all(["th", "td"], recursive=False)
if len(cells) == 2:
put(_txt(cells[0]), _txt(cells[1]))
for dl in soup.select("dl"):
dts, dds = dl.find_all("dt"), dl.find_all("dd")
if dts and len(dts) == len(dds):
for dt, dd in zip(dts, dds):
put(_txt(dt), _txt(dd))
for el in soup.find_all(True, class_=_LABEL_CLASS_RE):
sib = el.find_next_sibling(True)
if sib is not None and sib.get("class") and _VALUE_CLASS_RE.search(" ".join(sib.get("class"))):
put(_txt(el), _txt(sib))
# two-column rows (Bootstrap-style .row > .col + .col) and li "Label: value"
for row in soup.select("div[class*=row]"):
cols = [c for c in row.find_all(recursive=False) if c.name == "div" and any("col" in k for k in (c.get("class") or []))]
if len(cols) == 2:
l, v = _txt(cols[0]), _txt(cols[1])
if 0 < len(l) <= 40 and v:
put(l, v)
for li in soup.select("li, p, span"):
if li.find(True) is not None and li.name != "li":
continue
t = _txt(li)
if 3 < len(t) <= 160 and ":" in t:
l, v = t.split(":", 1)
if 1 < len(l.strip()) <= 40 and v.strip() and not l.strip()[0].isdigit():
put(l, v)
return specs
_MM_FIELDS = {"case_diameter_mm", "case_thickness"}
def apply_spec_map(specs: dict[str, str], spec_map: dict[str, str] | None) -> tuple[dict, dict]:
"""Split harvested specs into (promoted canonical fields, leftovers)."""
promoted, extra = {}, {}
smap = {_norm_label(k): v for k, v in (spec_map or {}).items()}
for label, value in specs.items():
col = smap.get(label)
if col:
promoted[col] = parse_mm(value) if col in _MM_FIELDS else value
else:
extra[label] = value
return promoted, extra
# ----------------------------------------------------------------------------- the parser
def parse_product_html(html: str, *, url: str = "", spec_map: dict | None = None,
sku: str | None = None) -> dict:
"""Return a dict of canonical fields + `extra_attributes` + `_payloads`
(JSON-LD / __NEXT_DATA__ raw, for hooks). Never raises on a thin page: a
page with no name and no price yields an almost-empty dict and the caller
decides whether that is a failure."""
soup = BeautifulSoup(html or "", "html.parser")
product, breadcrumb = extract_json_ld(html or "")
next_data = extract_next_data(html or "")
out: dict = {"product_url": url}
# --- JSON-LD
if product:
out["name"] = product.get("name") or ""
out["sku"] = product.get("sku") or product.get("mpn") or product.get("productID") or ""
b = product.get("brand")
if isinstance(b, dict):
out["_brand_name"] = b.get("name")
elif isinstance(b, str):
out["_brand_name"] = b
offers = product.get("offers") or {}
if isinstance(offers, list):
offers = offers[0] if offers else {}
if isinstance(offers, dict):
if offers.get("@type") == "AggregateOffer":
price = offers.get("lowPrice") or offers.get("price")
else:
price = offers.get("price")
pv, pc = parse_price(price)
out["price_value"] = pv
out["price_currency"] = offers.get("priceCurrency") or pc
out["stock_status"] = norm_availability(offers.get("availability"))
img = product.get("image")
if isinstance(img, dict):
img = img.get("url") or img.get("contentUrl")
if isinstance(img, list):
imgs = [i.get("url") if isinstance(i, dict) else i for i in img]
imgs = [i for i in imgs if i]
out["image_url"], out["image_count"] = (imgs[0], len(imgs)) if imgs else ("", 0)
elif isinstance(img, str):
out["image_url"], out["image_count"] = img, 1
out["description"] = product.get("description") or ""
agg = product.get("aggregateRating") or {}
if isinstance(agg, dict) and agg:
out["rating"] = agg.get("ratingValue")
out["review_count"] = agg.get("reviewCount") or agg.get("ratingCount")
if product.get("color"):
out["dial_color"] = product["color"] if isinstance(product["color"], str) else ""
for prop in product.get("additionalProperty") or []:
if isinstance(prop, dict) and prop.get("name"):
out.setdefault("_ld_props", {})[_norm_label(str(prop["name"]))] = str(prop.get("value") or "")
if product.get("audience", {}) and isinstance(product.get("audience"), dict):
out["gender"] = norm_gender(product["audience"].get("suggestedGender"))
# --- breadcrumb -> category (site-nav bucket)
if breadcrumb:
names = [e.get("name") or (e.get("item") or {}).get("name") if isinstance(e, dict) else None
for e in breadcrumb.get("itemListElement") or []]
names = [n for n in names if n]
if len(names) >= 2:
out["category"] = names[-2] if names[-1] == out.get("name") else names[-1]
# --- fallbacks from meta / DOM
if not out.get("name"):
og = soup.find("meta", property="og:title")
h1 = soup.find("h1")
out["name"] = (og.get("content") if og else None) or (_txt(h1) if h1 else "") or ""
if out.get("price_value") is None:
meta_p = soup.find("meta", property="product:price:amount") or soup.find("meta", itemprop="price")
if meta_p and meta_p.get("content"):
out["price_value"], cur = parse_price(meta_p["content"])
meta_c = soup.find("meta", property="product:price:currency") or soup.find("meta", itemprop="priceCurrency")
out["price_currency"] = (meta_c.get("content") if meta_c else None) or cur or out.get("price_currency")
else:
el = soup.find(attrs={"itemprop": "price"})
if el is not None:
out["price_value"], cur = parse_price(el.get("content") or _txt(el))
out["price_currency"] = out.get("price_currency") or cur
if not out.get("description"):
md = soup.find("meta", attrs={"name": "description"}) or soup.find("meta", property="og:description")
out["description"] = (md.get("content") if md else "") or ""
if not out.get("image_url"):
og = soup.find("meta", property="og:image")
if og and og.get("content"):
out["image_url"], out["image_count"] = og["content"], out.get("image_count") or 1
if sku and not out.get("sku"):
out["sku"] = sku
# --- specs
specs = harvest_specs(soup)
specs.update({k: v for k, v in (out.pop("_ld_props", {}) or {}).items() if k not in specs})
promoted, extra = apply_spec_map(specs, spec_map)
for k, v in promoted.items():
if v not in (None, ""):
out[k] = v
if "gender" in promoted:
out["gender"] = norm_gender(promoted["gender"]) or promoted["gender"]
if out.get("list_price_value") and out.get("price_value"):
try:
out["is_discounted"] = float(out["list_price_value"]) > float(out["price_value"])
except (TypeError, ValueError):
pass
out["extra_attributes"] = extra
out["_payloads"] = {"json_ld": product, "next_data": next_data, "specs": specs}
return out
def row_is_hollow(row: dict) -> bool:
"""A row with neither a name nor a price is not a product page we parsed."""
return not (row.get("name") or row.get("price_value"))
# ----------------------------------------------------------------------------- CLI
def _main(argv=None) -> int:
ap = argparse.ArgumentParser(description="Parse a saved product page into a canonical row.")
ap.add_argument("html_file")
ap.add_argument("--spec-map", default=None, help="JSON {label: canonical_field}")
ap.add_argument("--url", default="")
ap.add_argument("--labels", action="store_true", help="only list harvested spec labels")
ap.add_argument("--payloads", action="store_true", help="also print JSON-LD / __NEXT_DATA__ top-level keys")
a = ap.parse_args(argv)
html = Path(a.html_file).read_text(encoding="utf-8", errors="replace")
smap = json.loads(Path(a.spec_map).read_text(encoding="utf-8")) if a.spec_map else None
row = parse_product_html(html, url=a.url, spec_map=smap)
payloads = row.pop("_payloads")
if a.labels:
for k, v in payloads["specs"].items():
print(f"{k!r:40} -> {v[:80]!r}")
return 0
print(json.dumps(row, indent=2, ensure_ascii=False, default=str))
if a.payloads:
print("\n# json_ld keys:", sorted((payloads["json_ld"] or {}).keys()))
nd = payloads["next_data"] or {}
pp = ((nd.get("props") or {}).get("pageProps") or {}) if isinstance(nd, dict) else {}
print("# __NEXT_DATA__ props.pageProps keys:", sorted(pp.keys()) if isinstance(pp, dict) else None)
return 0
if __name__ == "__main__":
sys.exit(_main())