catalog-extract/scripts/run_catalog.py · py · download
#!/usr/bin/env python3
"""The product loop: urls.csv -> fetch each PDP through Nimble Extract -> parse ->
append-only CSV with snapshot_date.
python run_catalog.py --brand tissot --urls .nimble/tissot/urls.csv \
--render --wait 5000 --sentinel 'application/ld+json' \
--spec-map specs/tissot.json --out output/brands/tissot.csv --workers 6 --mode new
Guarantees (the engineering rules behind the Citizen pipeline):
- Every row is flushed as it is written; an interrupted run keeps everything.
- A fetch whose body fails the sentinel/JSON guard is retried, then counted
as a FAILURE. A parse that yields neither name nor price is a FAILURE.
Failures are never written as hollow rows.
- --exclude-name-regex marks accessories (straps, jewelry) as FILTERED, which
is not a failure and never counts toward the failure rate.
- Exit 1 only when >50% of products genuinely failed; a >15% catalog
contraction vs the previous snapshot is logged as VOLUME DROP (exit stays 0
unless --fail-on-drop).
- --mode new skips SKUs already in the CSV (nightly top-up); --mode full
rescrapes everything (price refresh).
Site-specific logic goes in a hook module (--hook path/to/brand_hook.py) that
defines any of:
def fetch(url, item, nx) -> str | None # replace the fetch entirely
def parse(html, item, row) -> dict | None | FILTERED # adjust/replace the parsed row
def prepare(nx) -> None # one-time setup (gender map, token read)
`row` arrives already parsed by parse_product; return it (modified), None for
"could not parse" (failure) or FILTERED to drop it on purpose.
"""
from __future__ import annotations
import argparse
import csv
import importlib.util
import json
import logging
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import nx # noqa: E402
from csvio import PRODUCT_FIELDS, IncrementalCsv, existing_keys, make_row, previous_snapshot_count # noqa: E402
from parse_product import parse_product_html, row_is_hollow # noqa: E402
import re # noqa: E402
logger = logging.getLogger("catalog")
FAILURE_RATE_LIMIT = 0.5
VOLUME_DROP_LIMIT = 0.15
class _Filtered:
def __repr__(self):
return "FILTERED"
def __bool__(self):
return False
FILTERED = _Filtered()
def load_hook(path: str | None):
if not path:
return None
spec = importlib.util.spec_from_file_location("brand_hook", path)
mod = importlib.util.module_from_spec(spec)
mod.FILTERED = FILTERED # so hooks can `return FILTERED`
spec.loader.exec_module(mod)
return mod
def read_urls(path: str) -> list[dict]:
with open(path, encoding="utf-8", newline="") as fh:
rows = list(csv.DictReader(fh))
for r in rows:
rj = r.get("record_json") or ""
r["record"] = json.loads(rj) if rj else None
return rows
def _one(item: dict, a, spec_map, hook, sentinel_re) -> tuple[str, dict | None]:
"""Returns (status, row) with status in {'ok','failed','filtered'}."""
url, sku = item["url"], item.get("sku") or ""
try:
if hook and hasattr(hook, "fetch"):
html = hook.fetch(url, item, nx)
else:
page = nx.fetch(url, render=a.render, wait_ms=a.wait, sentinel=sentinel_re,
driver=a.driver, country=a.country, request_timeout_ms=a.timeout,
tries=a.tries)
html = page.html if page.ok else None
if not html:
return "failed", None
row = parse_product_html(html, url=url, spec_map=spec_map, sku=sku or None)
payloads = row.pop("_payloads", None)
if a.exclude_name_regex and re.search(a.exclude_name_regex, row.get("name") or "", re.I):
return "filtered", None
if hook and hasattr(hook, "parse"):
row["_payloads"] = payloads
row["_record"] = item.get("record")
row = hook.parse(html, item, row)
if row is FILTERED:
return "filtered", None
if row is None:
return "failed", None
row.pop("_payloads", None)
row.pop("_record", None)
if row_is_hollow(row):
return "failed", None
row.pop("_brand_name", None)
if not row.get("sku"):
row["sku"] = sku
return "ok", row
except Exception as exc: # noqa: BLE001 - one product must never kill the run
logger.exception("unexpected error on %s: %s", url, exc)
return "failed", None
def _main(argv=None) -> int:
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--brand", required=True)
ap.add_argument("--urls", required=True, help="urls.csv from discover.py (sku,url,source,record_json)")
ap.add_argument("--out", required=True, help="append-only product CSV")
ap.add_argument("--render", action="store_true")
ap.add_argument("--wait", type=int, default=None, help="wait browser action in ms (4000-5000 for JS PDPs)")
ap.add_argument("--sentinel", default=None, help="substring (or re:REGEX) that proves the body arrived")
ap.add_argument("--spec-map", default=None, help="JSON {site label: canonical field}")
ap.add_argument("--hook", default=None, help="python file with fetch()/parse()/prepare()")
ap.add_argument("--exclude-name-regex", default=None, help="names matching this are FILTERED (not failures)")
ap.add_argument("--mode", choices=["new", "full"], default="new")
ap.add_argument("--limit", type=int, default=None)
ap.add_argument("--workers", type=int, default=6)
ap.add_argument("--tries", type=int, default=3)
ap.add_argument("--driver", default=nx.DEFAULT_DRIVER)
ap.add_argument("--country", default="US")
ap.add_argument("--timeout", type=int, default=150_000)
ap.add_argument("--fail-on-drop", action="store_true")
ap.add_argument("--log", default=None, help="log file (default: logs/<brand>_<ts>.log)")
a = ap.parse_args(argv)
log_path = Path(a.log or f"logs/{a.brand}_{time.strftime('%Y%m%d_%H%M%S')}.log")
log_path.parent.mkdir(parents=True, exist_ok=True)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s",
handlers=[logging.StreamHandler(), logging.FileHandler(log_path, encoding="utf-8")])
spec_map = json.loads(Path(a.spec_map).read_text(encoding="utf-8")) if a.spec_map else None
hook = load_hook(a.hook)
sentinel_re = re.compile(a.sentinel[3:]) if a.sentinel and a.sentinel.startswith("re:") else a.sentinel
items = read_urls(a.urls)
if not items:
logger.error("no URLs in %s — discovery is dead; aborting", a.urls)
return 1
logger.info("=== %s: %d discovered, mode=%s workers=%d render=%s wait=%s ===",
a.brand, len(items), a.mode, a.workers, a.render, a.wait)
prev_date, prev_count = previous_snapshot_count(a.out)
skipped = 0
if a.mode == "new":
known = existing_keys(a.out)
before = len(items)
items = [i for i in items if (i.get("sku") or "").strip() not in known]
skipped = before - len(items)
logger.info("mode=new: %d already recorded, %d to scrape", skipped, len(items))
seen, dedup = set(), []
for i in items:
key = (i.get("sku") or "").strip() or i["url"]
if key in seen:
continue
seen.add(key)
dedup.append(i)
items = dedup[: a.limit] if a.limit else dedup
if not items:
logger.info("nothing to scrape")
return 0
if hook and hasattr(hook, "prepare"):
hook.prepare(nx)
ok = failed = filtered = 0
failed_urls: list[str] = []
t0 = time.time()
with IncrementalCsv(a.out, PRODUCT_FIELDS) as csvw, ThreadPoolExecutor(max_workers=max(1, a.workers)) as ex:
futs = {ex.submit(_one, it, a, spec_map, hook, sentinel_re): it for it in items}
for n, fut in enumerate(as_completed(futs), 1):
it = futs[fut]
status, row = fut.result()
if status == "ok":
csvw.write(make_row(PRODUCT_FIELDS, a.brand, row))
ok += 1
elif status == "filtered":
filtered += 1
else:
failed += 1
failed_urls.append(it["url"])
if n % 25 == 0 or n == len(items):
logger.info("progress %d/%d ok=%d failed=%d filtered=%d (%.0fs)",
n, len(items), ok, failed, filtered, time.time() - t0)
exit_code = 0
attempted = ok + failed
if attempted and failed / attempted > FAILURE_RATE_LIMIT:
logger.error("FAILURE RATE %.0f%% (%d/%d) exceeds %.0f%% — exit 1",
100 * failed / attempted, failed, attempted, 100 * FAILURE_RATE_LIMIT)
exit_code = 1
volume_drop = False
if a.mode == "full" and prev_count and ok < prev_count * (1 - VOLUME_DROP_LIMIT):
volume_drop = True
logger.error("VOLUME DROP: %d rows this run vs %d on %s (%.0f%% contraction)",
ok, prev_count, prev_date, 100 * (1 - ok / prev_count))
if a.fail_on_drop:
exit_code = 1
if failed_urls:
fp = log_path.with_suffix(".failed.txt")
fp.write_text("\n".join(failed_urls), encoding="utf-8")
logger.info("failed URLs written to %s", fp)
logger.info("=== DONE %s ok=%d failed=%d filtered=%d skipped=%d%s billed_requests=%d exit=%d out=%s ===",
a.brand, ok, failed, filtered, skipped, " VOLUME_DROP" if volume_drop else "",
nx.billed_requests(), exit_code, a.out)
return exit_code
if __name__ == "__main__":
sys.exit(_main())