catalog-extract/hooks/example_hook.py · py · download
"""Example hook for run_catalog.py --hook.
A hook holds the site-specific part of a brand scraper: anything the generic
parser (JSON-LD + spec harvest + spec map) cannot reach. Define any of:
prepare(nx) one-time setup on the main thread (gender map,
a public token read from the page). Worker
threads must not build shared state lazily.
fetch(url, item, nx) replace the fetch (custom sentinel/wait/cookies);
return HTML or None.
parse(html, item, row) adjust/replace the parsed row; return the row,
None (= failure, counted) or FILTERED (= on
purpose, not counted).
`item` is the urls.csv row ({sku,url,source,record}); `row["_payloads"]` has
{"json_ld": {...}, "next_data": {...}, "specs": {label: value}}; `row["_record"]`
is the Shopify product JSON when discovery came from products.json.
`FILTERED` is injected into the module namespace by run_catalog.
"""
from __future__ import annotations
import re
GENDER_MAP: dict[str, str] = {}
_SPEC_ITEM_SENTINEL = "product-specification" # a class only a COMPLETE PDP contains
_NOT_A_WATCH = re.compile(r"\b(strap|bracelet only|jewel|bag|wallet|gift card|charger)\b", re.I)
def prepare(nx):
"""Build {sku: gender} from the men's / women's landing pages once."""
for gender, url in (("men's", "https://www.brand.com/en-us/men"),
("women's", "https://www.brand.com/en-us/women")):
page = nx.fetch(url, render=True, wait_ms=5000, sentinel="product-tile")
for sku in re.findall(r"/en-us/([A-Z]\d{8,})\.html", page.html or ""):
GENDER_MAP.setdefault(sku, gender)
def fetch(url, item, nx):
page = nx.fetch(url, render=True, wait_ms=5000, sentinel=_SPEC_ITEM_SENTINEL,
request_timeout_ms=150_000)
return page.html if page.ok else None
def parse(html, item, row):
if _NOT_A_WATCH.search(row.get("name") or ""):
return FILTERED # noqa: F821 - injected by run_catalog
row["gender"] = row.get("gender") or GENDER_MAP.get(row.get("sku") or item.get("sku") or "", "")
# Example: pull specs out of __NEXT_DATA__ when the DOM has none.
nd = (row.get("_payloads") or {}).get("next_data") or {}
product = (((nd.get("props") or {}).get("pageProps") or {}).get("product") or {})
if product:
row["collection"] = row.get("collection") or product.get("collectionName") or ""
row["list_price_value"] = row.get("list_price_value") or product.get("compareAtPrice")
# Example: a Shopify record carries price + compare_at_price per variant.
rec = row.get("_record") or {}
variants = rec.get("variants") or []
if variants and not row.get("price_value"):
row["price_value"] = float(variants[0].get("price") or 0) or None
row["price_currency"] = row.get("price_currency") or "USD"
if not row.get("movement_type") and not row.get("case_material"):
return None # spec block never arrived: a counted failure, not a hollow row
return row