catalog-extract/scripts/discover.py · py · download
#!/usr/bin/env python3
"""Enumerate every product URL for a brand from a site-owned source -> urls.csv.
Discovery must come from the site (sitemap, products.json, grid endpoint,
paginated listing), never a hardcoded list: a static list silently misses
additions, and volume tripwires only catch drops.
Modes (pick the first that works for the site, in this order):
--sitemap URL XML sitemap or sitemap index (indexes are followed).
Filter product URLs with --match REGEX; pull the SKU
with --sku-regex REGEX (first group).
--shopify BASE Shopify: BASE/products.json?limit=250&page=N until a
short page. Each row carries the product JSON (price,
variants, tags) in `record` so PDP fetches are optional.
--collection HANDLE scopes to one collection.
--listing TEMPLATE Paginated listing pages; TEMPLATE contains {page}.
Product links are found with --link-regex (applied to
the raw HTML; first group = URL). Stops when a page adds
nothing new or --max-pages is reached. --render for
JS-rendered grids (+ --wait ms).
Output CSV columns: sku,url,source,record_json
python discover.py --sitemap https://www.tissotwatches.com/en-us/sitemap-custom-sitemap_0-product-en_US.xml \
--match '/en-us/T\\d+\\.html$' --sku-regex '/en-us/(T\\d+)\\.html' --out .nimble/tissot/urls.csv
python discover.py --shopify https://www.timex.com --out .nimble/timex/urls.csv
python discover.py --listing 'https://www.macys.com/shop/featured/tissot-watches/Pageindex/{page}' \
--link-regex 'href="(/shop/product/[^"?]+\\?ID=\\d+)' --base https://www.macys.com --out urls.csv
"""
from __future__ import annotations
import argparse
import csv
import json
import logging
import re
import sys
from pathlib import Path
from urllib.parse import urljoin
sys.path.insert(0, str(Path(__file__).resolve().parent))
import nx # noqa: E402
logger = logging.getLogger("discover")
_LOC_RE = re.compile(r"<loc>\s*([^<\s]+)\s*</loc>", re.I)
_SITEMAP_RE = re.compile(r"<sitemap>.*?<loc>\s*([^<\s]+)\s*</loc>.*?</sitemap>", re.I | re.S)
def from_sitemap(url: str, *, match: str | None, sku_regex: str | None, max_sitemaps: int = 50,
**fetch_kw) -> list[dict]:
seen_maps, queue, out = set(), [url], []
m_re = re.compile(match) if match else None
s_re = re.compile(sku_regex) if sku_regex else None
while queue and len(seen_maps) < max_sitemaps:
u = queue.pop(0)
if u in seen_maps:
continue
seen_maps.add(u)
page = nx.fetch(u, render=False, sentinel="<loc>", **fetch_kw)
if not page.ok:
logger.warning("sitemap fetch failed: %s", u)
continue
children = _SITEMAP_RE.findall(page.html)
if children:
logger.info("%s is an index with %d child sitemaps", u, len(children))
queue.extend(c for c in children if (m_re is None or not m_re.search(c)))
# child sitemap names often say "product"; walk them all but log it
continue
locs = _LOC_RE.findall(page.html)
kept = 0
for loc in locs:
if m_re and not m_re.search(loc):
continue
sku = (s_re.search(loc).group(1) if s_re and s_re.search(loc) else "")
out.append({"sku": sku, "url": loc, "source": u, "record_json": ""})
kept += 1
logger.info("%s: %d <loc>, %d kept", u, len(locs), kept)
return out
def from_shopify(base: str, *, collection: str | None = None, page_cap: int = 200, **fetch_kw) -> list[dict]:
base = base.rstrip("/")
out = []
for page in range(1, page_cap + 1):
path = f"/collections/{collection}/products.json" if collection else "/products.json"
url = f"{base}{path}?limit=250&page={page}"
data = nx.fetch_json(url, wait_ms=8000, request_timeout_ms=180_000, **fetch_kw)
if not isinstance(data, dict):
logger.error("products.json page %d unparseable after retries (truncation?) — stopping", page)
break
products = data.get("products") or []
for p in products:
handle = p.get("handle")
variants = p.get("variants") or []
sku = (variants[0].get("sku") if variants else "") or str(p.get("id") or "")
out.append({"sku": sku, "url": f"{base}/products/{handle}", "source": url,
"record_json": json.dumps(p, ensure_ascii=False)})
logger.info("products.json page %d -> %d products (total %d)", page, len(products), len(out))
if len(products) < 250:
break
return out
def from_listing(template: str, *, link_regex: str, base: str | None, max_pages: int,
render: bool, wait_ms: int | None, sentinel: str | None, sku_regex: str | None,
**fetch_kw) -> list[dict]:
l_re, s_re = re.compile(link_regex), (re.compile(sku_regex) if sku_regex else None)
out, seen = [], set()
for page in range(1, max_pages + 1):
url = template.format(page=page)
p = nx.fetch(url, render=render, wait_ms=wait_ms, sentinel=sentinel, **fetch_kw)
if not p.ok:
logger.warning("listing page %d failed; stopping", page)
break
links = [m.group(1) for m in l_re.finditer(p.html)]
new = 0
for href in links:
full = urljoin(base or url, href.replace("&", "&"))
if full in seen:
continue
seen.add(full)
sku = (s_re.search(full).group(1) if s_re and s_re.search(full) else "")
out.append({"sku": sku, "url": full, "source": url, "record_json": ""})
new += 1
logger.info("page %d: %d links, %d new (total %d)", page, len(links), new, len(out))
if new == 0:
break
return out
def write_urls(rows: list[dict], out: str) -> int:
# de-dupe by sku when present, else by url
seen, kept = set(), []
for r in rows:
key = r["sku"] or r["url"]
if key in seen:
continue
seen.add(key)
kept.append(r)
Path(out).parent.mkdir(parents=True, exist_ok=True)
with open(out, "w", encoding="utf-8", newline="") as fh:
w = csv.DictWriter(fh, fieldnames=["sku", "url", "source", "record_json"])
w.writeheader()
w.writerows(kept)
return len(kept)
def _main(argv=None) -> int:
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
g = ap.add_mutually_exclusive_group(required=True)
g.add_argument("--sitemap")
g.add_argument("--shopify")
g.add_argument("--listing")
ap.add_argument("--match", help="regex a product URL must match (sitemap mode)")
ap.add_argument("--sku-regex", help="regex with one group capturing the SKU from the URL")
ap.add_argument("--collection", help="Shopify collection handle")
ap.add_argument("--link-regex", help="regex with one group capturing product hrefs (listing mode)")
ap.add_argument("--base", help="base URL for relative hrefs (listing mode)")
ap.add_argument("--max-pages", type=int, default=60)
ap.add_argument("--render", action="store_true")
ap.add_argument("--wait", type=int, default=None)
ap.add_argument("--sentinel", default=None)
ap.add_argument("--country", default="US")
ap.add_argument("--driver", default=nx.DEFAULT_DRIVER)
ap.add_argument("--out", required=True)
a = ap.parse_args(argv)
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
fk = {"country": a.country, "driver": a.driver}
if a.sitemap:
rows = from_sitemap(a.sitemap, match=a.match, sku_regex=a.sku_regex, **fk)
elif a.shopify:
rows = from_shopify(a.shopify, collection=a.collection, **fk)
else:
if not a.link_regex:
ap.error("--listing needs --link-regex")
rows = from_listing(a.listing, link_regex=a.link_regex, base=a.base, max_pages=a.max_pages,
render=a.render, wait_ms=a.wait, sentinel=a.sentinel, sku_regex=a.sku_regex, **fk)
n = write_urls(rows, a.out)
missing_sku = sum(1 for r in rows if not r["sku"])
print(json.dumps({"urls": n, "without_sku": missing_sku, "out": a.out,
"billed_requests": nx.billed_requests()}))
return 0 if n else 1
if __name__ == "__main__":
sys.exit(_main())