catalog-extract/scripts/recon.py · py · download
#!/usr/bin/env python3
"""Reconnaissance for a new brand or retailer site, through Nimble Extract only.
Answers the questions that decide how the scraper is built:
1. Discovery source: robots.txt sitemaps (and which child looks like products),
Shopify products.json, Salesforce Commerce Cloud grid, Magento.
2. Render requirement for a PDP: is the price / JSON-LD / spec block present in
the raw server HTML (render=False) or only after JS (render=True)?
3. Where the data lives: JSON-LD Product, __NEXT_DATA__, inline state objects,
and which spec labels the page prints (for the spec map).
Writes a markdown report + the raw pages under .nimble/recon/<host>/.
python recon.py https://www.hamiltonwatch.com --pdp https://www.hamiltonwatch.com/en-us/h32475140-jazzmaster-auto.html
"""
from __future__ import annotations
import argparse
import json
import logging
import re
import sys
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from urllib.parse import urlparse
sys.path.insert(0, str(Path(__file__).resolve().parent))
import nx # noqa: E402
from parse_product import extract_json_ld, extract_next_data, harvest_specs, parse_product_html # noqa: E402
from bs4 import BeautifulSoup # noqa: E402
logger = logging.getLogger("recon")
_LOC_RE = re.compile(r"<loc>\s*([^<\s]+)\s*</loc>", re.I)
def _platform_hints(html: str) -> list[str]:
hints = []
checks = {
"Shopify": r"cdn\.shopify\.com|Shopify\.theme|/products\.json",
"Salesforce Commerce Cloud (SFCC)": r"demandware\.static|/on/demandware\.store|Search-UpdateGrid",
"Magento": r"Magento_|mage/|/static/version\d+/|requirejs-config",
"Next.js": r"__NEXT_DATA__|/_next/static",
"Nuxt": r"__NUXT__|/_nuxt/",
"Algolia": r"algolia(net|\.com)|algoliasearch|InstantSearch",
"Akamai edge": r"akam/|_abck|bm_sz",
"Cloudflare": r"cdn-cgi/challenge|__cf_bm",
"Where2GetIt/Brandify locator": r"where2getit|brandify",
"Bazaarvoice reviews": r"bazaarvoice|bvapi",
"Yotpo reviews": r"yotpo",
}
for name, pat in checks.items():
if re.search(pat, html, re.I):
hints.append(name)
return hints
def recon_site(base: str, out_dir: Path, country: str, driver: str) -> dict:
base = base.rstrip("/")
rep: dict = {"base": base, "sitemaps": [], "product_sitemaps": [], "shopify": None, "robots": None}
robots = nx.fetch(f"{base}/robots.txt", render=False, country=country, driver=driver, tries=1)
if robots.ok:
txt = robots.html
if "<pre" in txt[:500]:
txt = re.sub(r"<[^>]+>", "", txt)
rep["robots"] = txt[:3000]
rep["sitemaps"] = re.findall(r"(?im)^\s*sitemap:\s*(\S+)", txt)
(out_dir / "robots.txt").write_text(txt, encoding="utf-8")
candidates = rep["sitemaps"] or [f"{base}/sitemap.xml", f"{base}/sitemap_index.xml"]
def _sm(sm):
p = nx.fetch(sm, render=False, sentinel="<loc>", country=country, driver=driver, tries=1)
if not p.ok:
return {"url": sm, "error": p.error}
locs = _LOC_RE.findall(p.html)
return {"url": sm, "index": "<sitemapindex" in p.html[:2000], "entries": len(locs), "sample": locs[:3], "locs": locs}
with ThreadPoolExecutor(max_workers=8) as ex:
first = list(ex.map(_sm, candidates[:5]))
children = []
for e in first:
if "error" in e:
rep.setdefault("sitemap_errors", []).append(e["url"])
continue
rep["product_sitemaps"].append({k: v for k, v in e.items() if k != "locs"})
if e["index"]:
# match on the path only (the host often contains "watch"); prefer the US/English locale
kids = [c for c in e["locs"] if re.search(r"product|pdp|catalog|item|en[-_]us", urlparse(c).path, re.I)]
kids.sort(key=lambda c: (0 if re.search(r"en[-_]us", c, re.I) else 1))
rep.setdefault("index_children", []).extend(e["locs"][:40])
children.extend(kids[:6]) # cap: the point is to find the product sitemap, not walk them all
for e in ex.map(_sm, children):
if "error" not in e:
rep["product_sitemaps"].append({k: v for k, v in e.items() if k != "locs"})
shopify_f = ex.submit(nx.fetch_json, f"{base}/products.json?limit=1", wait_ms=3000, country=country, driver=driver, tries=1)
home_f = ex.submit(nx.fetch, base + "/", render=False, country=country, driver=driver, tries=1)
sj, home = shopify_f.result(), home_f.result()
if isinstance(sj, dict) and isinstance(sj.get("products"), list):
rep["shopify"] = {"products_json": True, "sample_handle": (sj["products"][0].get("handle") if sj["products"] else None)}
rep["home_bytes_raw"] = home.size
rep["platform_hints"] = _platform_hints(home.html) if home.ok else []
if home.ok:
(out_dir / "home_raw.html").write_text(home.html, encoding="utf-8")
return rep
def recon_pdp(url: str, out_dir: Path, country: str, driver: str, wait_ms: int) -> dict:
rep: dict = {"url": url}
with ThreadPoolExecutor(max_workers=2) as ex: # raw and rendered in parallel
pages = dict(zip(("raw", "rendered"), ex.map(
lambda render: nx.fetch(url, render=render, wait_ms=wait_ms if render else None, country=country,
driver=driver, tries=2, request_timeout_ms=150_000),
(False, True))))
for mode in ("raw", "rendered"):
p = pages[mode]
info: dict = {"ok": p.ok, "bytes": p.size, "error": p.error}
if p.ok:
html = p.html
(out_dir / f"pdp_{mode}.html").write_text(html, encoding="utf-8")
ld, _ = extract_json_ld(html)
nd = extract_next_data(html)
soup = BeautifulSoup(html, "html.parser")
specs = harvest_specs(soup)
row = parse_product_html(html, url=url)
info.update({
"json_ld_product": bool(ld),
"json_ld_keys": sorted(ld.keys())[:25] if ld else [],
"next_data": nd is not None,
"next_pageprops_keys": sorted(((nd or {}).get("props") or {}).get("pageProps", {}).keys())[:25] if nd else [],
"inline_state_markers": [m for m in ("__PRELOADED_STATE__", "__INITIAL_STATE__", "dataLayer.push", "InstantSearchInitialResults", "window.__NUXT__", "digitalData") if m in html],
"price_in_html": bool(re.search(r'"price"\s*:\s*"?\d|itemprop="price"|product:price:amount', html)),
"name_parsed": row.get("name"), "price_parsed": row.get("price_value"),
"spec_labels": list(specs.keys())[:60], "spec_label_count": len(specs),
"platform_hints": _platform_hints(html),
})
rep[mode] = info
raw, ren = rep.get("raw", {}), rep.get("rendered", {})
def score(i):
return (bool(i.get("json_ld_product")) + bool(i.get("next_data")) + bool(i.get("price_parsed"))
+ min(i.get("spec_label_count", 0), 10) / 10)
if raw.get("ok") and score(raw) >= score(ren) - 0.05 and (raw.get("price_parsed") or raw.get("json_ld_product") or raw.get("next_data")):
rep["recommendation"] = "render=False (server-rendered: cheap path)"
elif ren.get("ok"):
rep["recommendation"] = f"render=True + wait {wait_ms}ms (client-hydrated)"
else:
rep["recommendation"] = "UNVERIFIED - both fetches failed; try another driver or a wait, then a cookie warm-up"
return rep
def _md(site: dict | None, pdp: dict | None) -> str:
L = ["# Recon report", ""]
if site:
L += [f"## Site: {site['base']}", "",
f"- platform hints: {', '.join(site.get('platform_hints') or []) or 'none'}",
f"- robots sitemaps: {site.get('sitemaps') or 'none listed'}",
f"- Shopify products.json: {site.get('shopify') or 'no'}", ""]
for s in site.get("product_sitemaps") or []:
L.append(f"- sitemap {s['url']} ({'index' if s['index'] else 'urlset'}, {s['entries']} entries) sample: {s['sample'][:2]}")
if site.get("index_children"):
L += ["", "Index children (first 40; only product-looking ones were opened):"]
L += [f" - {c}" for c in site["index_children"]]
L.append("")
if pdp:
L += [f"## PDP: {pdp['url']}", "", "| mode | ok | bytes | JSON-LD | __NEXT_DATA__ | price parsed | spec labels |", "|---|---|---|---|---|---|---|"]
for mode in ("raw", "rendered"):
i = pdp.get(mode, {})
L.append(f"| {mode} | {i.get('ok')} | {i.get('bytes')} | {i.get('json_ld_product')} | {i.get('next_data')} | {i.get('price_parsed')} | {i.get('spec_label_count')} |")
L += ["", f"**Recommendation:** {pdp.get('recommendation')}", ""]
best = pdp.get("raw") if "render=False" in pdp.get("recommendation", "") else pdp.get("rendered")
if best:
L += [f"- parsed name: {best.get('name_parsed')!r}", f"- inline state: {best.get('inline_state_markers')}",
f"- JSON-LD keys: {best.get('json_ld_keys')}", f"- __NEXT_DATA__ pageProps keys: {best.get('next_pageprops_keys')}",
"", "Spec labels seen (feed these into the spec map):", ""]
L += [f"- `{s}`" for s in best.get("spec_labels") or []]
L += ["", f"billed requests: {nx.billed_requests()}"]
return "\n".join(L)
def _main(argv=None) -> int:
ap = argparse.ArgumentParser(description="Recon a site + one PDP through Nimble Extract.")
ap.add_argument("base", nargs="?", help="https://www.brand.com (site-level recon)")
ap.add_argument("--pdp", help="one product URL to test raw vs rendered")
ap.add_argument("--wait", type=int, default=5000)
ap.add_argument("--country", default="US")
ap.add_argument("--driver", default=nx.DEFAULT_DRIVER)
ap.add_argument("--out-dir", default=None)
a = ap.parse_args(argv)
if not a.base and not a.pdp:
ap.error("give a base URL and/or --pdp")
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
host = urlparse(a.base or a.pdp).netloc
out_dir = Path(a.out_dir or f".nimble/recon/{host}")
out_dir.mkdir(parents=True, exist_ok=True)
site = recon_site(a.base, out_dir, a.country, a.driver) if a.base else None
pdp = recon_pdp(a.pdp, out_dir, a.country, a.driver, a.wait) if a.pdp else None
(out_dir / "recon.json").write_text(json.dumps({"site": site, "pdp": pdp}, indent=2, default=str), encoding="utf-8")
md = _md(site, pdp)
(out_dir / "recon.md").write_text(md, encoding="utf-8")
print(md)
print(f"\nsaved: {out_dir}/recon.md (+ raw pages)")
return 0
if __name__ == "__main__":
sys.exit(_main())