#!/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/_.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())