catalog-extract/scripts/stores.py · py · download
#!/usr/bin/env python3
"""Store locators through Nimble Extract -> append-only stores CSV.
Four mechanisms cover every locator seen so far; try them in this order:
--json-api URL the site's own locator returns JSON to a plain GET
(Grand Seiko / Seiko /api/v2/stores). Fetched with
render=False + 8s wait, <pre> unwrapped, parse-validated.
--browser-api URL the API needs a POST body or exact headers (Magento
--page PAGE_URL REST XHR with Accept: application/json, an Algolia POST
[--method POST] with X-Algolia-* headers, a Where2GetIt widget POST).
[--headers JSON] The page is loaded in the browser and the call is made
[--body JSON|string] from that origin via a `fetch` browser action.
--grid points.csv same as either of the above, repeated per lat/lng point
(URL/body may contain {lat} {lng}); for locators that cap
at N-nearest with no pagination (Movado: 200). Rows are
de-duplicated on --dedup-key. Counts are a FLOOR.
SSR country pages (Tudor) are a discovery + parse job: use discover.py
--listing with a link regex, then run_catalog.py with a
hook that maps the detail page to STORE_FIELDS.
Mapping the response to the canonical schema:
--records PATH dotted path to the list of store objects
("stores", "items", "hits", "response.collection")
--map JSON {canonical_field: source_key_or_dotted_path}, e.g.
{"store_name":"name","city":"city.0.name","latitude":"lat"}
--country-filter US keep only rows whose country_code equals this
--const JSON constant fields, e.g. {"country_code":"US","store_type":"Authorized Dealer"}
Never fabricate rows: if the locator returns nothing parseable, the run logs a
WARNING, dumps the raw body under .nimble/stores/, and writes 0 rows.
Public front-end tokens (an Algolia search key, a widget appkey) are read live
from the locator page at runtime (--token-from-page REGEX with one group, then
use {token} in URL/headers/body). Never hardcode them in the command history.
"""
from __future__ import annotations
import argparse
import csv
import json
import logging
import re
import sys
import time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import nx # noqa: E402
from csvio import STORE_FIELDS, IncrementalCsv, make_row # noqa: E402
logger = logging.getLogger("stores")
def dig(obj, path: str):
"""'a.b.0.c' -> obj['a']['b'][0]['c'] (None when any hop is missing)."""
cur = obj
for hop in (path or "").split("."):
if hop == "":
continue
if isinstance(cur, list):
try:
cur = cur[int(hop)]
except (ValueError, IndexError):
return None
elif isinstance(cur, dict):
cur = cur.get(hop)
else:
return None
if cur is None:
return None
return cur
def map_record(rec: dict, fmap: dict[str, str], const: dict) -> dict:
row = dict(const)
for canon, src in fmap.items():
v = dig(rec, src)
if isinstance(v, str):
v = re.sub(r"\s*[\r\n]+\s*", ", ", v.strip())
row[canon] = v
return row
def _sub(template: str, **kw) -> str:
out = template
for k, v in kw.items():
out = out.replace("{" + k + "}", str(v))
return out
def fetch_records(a, token: str | None, lat=None, lng=None) -> list[dict] | None:
kw = dict(lat=lat if lat is not None else "", lng=lng if lng is not None else "", token=token or "")
if a.json_api:
url = _sub(a.json_api, **kw)
data = nx.fetch_json(url, wait_ms=a.wait, country=a.country, driver=a.driver, request_timeout_ms=a.timeout)
else:
headers = json.loads(_sub(a.headers, **kw)) if a.headers else None
body = _sub(a.body, **kw) if a.body else None
req = {"url": _sub(a.browser_api, **kw), "method": a.method}
if headers:
req["headers"] = headers
if body is not None:
req["body"] = body
res = nx.browser_fetch(a.page, [req], wait_ms=a.wait or 4000, country=a.country, driver=a.driver,
request_timeout_ms=max(a.timeout, 240_000))
data = res[0] if res else None
if data is None:
return None
recs = dig(data, a.records) if a.records else data
if isinstance(recs, dict):
recs = list(recs.values())
if not isinstance(recs, list):
dump = Path(".nimble/stores") / f"{a.brand}_unparsed_{int(time.time())}.json"
dump.parent.mkdir(parents=True, exist_ok=True)
dump.write_text(json.dumps(data, indent=2, default=str)[:2_000_000], encoding="utf-8")
logger.warning("records path %r not found in response; dumped %s (top keys: %s)",
a.records, dump, list(data.keys())[:20] if isinstance(data, dict) else type(data))
return None
return [r for r in recs if isinstance(r, dict)]
def _main(argv=None) -> int:
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--brand", required=True)
ap.add_argument("--out", required=True)
src = ap.add_mutually_exclusive_group(required=True)
src.add_argument("--json-api")
src.add_argument("--browser-api")
ap.add_argument("--page", help="page to load before an in-browser fetch (same origin as the API ideally)")
ap.add_argument("--method", default="GET")
ap.add_argument("--headers", default=None, help="JSON object")
ap.add_argument("--body", default=None, help="JSON string or raw body")
ap.add_argument("--records", default=None, help="dotted path to the list of stores")
ap.add_argument("--map", required=True, help="JSON {canonical_field: source path}")
ap.add_argument("--const", default=None, help="JSON constant fields")
ap.add_argument("--country-filter", default=None)
ap.add_argument("--grid", default=None, help="CSV with lat,lng columns")
ap.add_argument("--dedup-key", default="store_name,address", help="comma list of canonical fields")
ap.add_argument("--token-from-page", default=None, help="regex (1 group) read from --page HTML; use {token}")
ap.add_argument("--wait", type=int, default=8000)
ap.add_argument("--country", default="US")
ap.add_argument("--driver", default=nx.DEFAULT_DRIVER)
ap.add_argument("--timeout", type=int, default=180_000)
ap.add_argument("--limit-points", type=int, default=None)
a = ap.parse_args(argv)
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
if a.browser_api and not a.page:
ap.error("--browser-api needs --page")
fmap = json.loads(a.map)
const = json.loads(a.const) if a.const else {}
token = None
if a.token_from_page:
p = nx.fetch(a.page or a.json_api, render=True, wait_ms=6000, country=a.country, driver=a.driver)
m = re.search(a.token_from_page, p.html or "")
if not m:
logger.error("token regex did not match the page; refusing to guess")
return 1
token = m.group(1)
logger.info("read a %d-char token from the page", len(token))
points = [(None, None)]
if a.grid:
with open(a.grid, encoding="utf-8", newline="") as fh:
points = [(r["lat"], r["lng"]) for r in csv.DictReader(fh)]
if a.limit_points:
points = points[: a.limit_points]
logger.info("grid sweep over %d points (counts are a floor under an N-nearest cap)", len(points))
keys = [k.strip() for k in a.dedup_key.split(",") if k.strip()]
seen, written, empty_points = set(), 0, 0
with IncrementalCsv(a.out, STORE_FIELDS) as w:
for lat, lng in points:
recs = fetch_records(a, token, lat, lng)
if recs is None:
empty_points += 1
continue
for rec in recs:
row = map_record(rec, fmap, const)
if a.country_filter and str(row.get("country_code") or "").upper() != a.country_filter.upper():
continue
k = tuple(str(row.get(x) or "").strip().lower() for x in keys)
if any(k) and k in seen:
continue
seen.add(k)
w.write(make_row(STORE_FIELDS, a.brand, row, extra_key=None))
written += 1
logger.info("point (%s,%s): %d records, %d written so far", lat, lng, len(recs), written)
if written == 0:
logger.warning("0 store rows written — locator unverified; nothing fabricated.")
print(json.dumps({"brand": a.brand, "stores": written, "points": len(points), "empty_points": empty_points,
"billed_requests": nx.billed_requests(), "out": a.out}))
return 0 if written else 1
if __name__ == "__main__":
sys.exit(_main())