#!/usr/bin/env python3 """Build catalog-with-details.json from the 604-family import file + KIKO pages. Usage: cd admin python3 build_catalog_with_details.py python3 build_catalog_with_details.py --limit 20 python3 build_catalog_with_details.py --resume """ from __future__ import annotations import argparse import json import re import sys import time from pathlib import Path ROOT = Path(__file__).resolve().parent DIST_DATA = ROOT.parent / "dist" / "data" SRC = DIST_DATA / "catalog-import-ready-families.json" OUT = DIST_DATA / "catalog-with-details.json" STATE = DIST_DATA / "catalog-with-details.state.json" sys.path.insert(0, str(ROOT)) from detail_schema import ensure_product_queries # noqa: E402 from kiko_enrich import ( # noqa: E402 _family_key_from_product_url, discover_catalog_product_urls, discover_product_url, fetch_kiko_product, ) def _norm_tokens(text: str) -> str: return re.sub(r"[^a-z0-9]+", "-", (text or "").lower()).strip("-") def family_candidates(product: dict) -> list[str]: out: list[str] = [] slug = (product.get("slug") or "").strip().lower() if slug: out.append(slug) m = re.match(r"^(.+)-(\d+)-(\d+)$", slug) if m: out.append(m.group(1)) m = re.match(r"^(.+)-(\d{1,3})[a-z]{0,4}$", slug) if m: out.append(m.group(1)) out.append(re.sub(r"-\d+$", "", slug)) name = product.get("name") or "" name_base = re.sub(r"\s+\d+[A-Za-z]{0,4}$", "", name).strip() out.append(_norm_tokens(name_base)) out.append(_norm_tokens(name)) seen = set() uniq = [] for k in out: if k and k not in seen: seen.add(k) uniq.append(k) return uniq def load_json(path: Path): return json.loads(path.read_text(encoding="utf-8")) def save_json(path: Path, data) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") tmp.replace(path) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--limit", type=int, default=0, help="Max products to enrich (0=all)") ap.add_argument("--resume", action="store_true") ap.add_argument("--market", default="en-gb") ap.add_argument("--sleep", type=float, default=0.4) args = ap.parse_args() src = load_json(SRC) products = list(src.get("products") or []) if args.resume and OUT.exists(): prev = load_json(OUT) by_id = {p.get("id"): p for p in prev.get("products") or [] if p.get("id")} products = [by_id.get(p.get("id"), p) for p in products] print(f"Resumed from {OUT} ({len(by_id)} prior rows)") print("Loading KIKO sitemap families…") discovered = discover_catalog_product_urls(args.market, one_per_family=True) by_fam = {_family_key_from_product_url(u): u for u in discovered["urls"]} print(f"KIKO families: {len(by_fam)}") state = {"ok": 0, "fail": 0, "skip": 0, "errors": []} if STATE.exists() and args.resume: try: state.update(load_json(STATE)) except Exception: pass limit = args.limit if args.limit > 0 else len(products) enriched = 0 for i, product in enumerate(products): if enriched >= limit: break product = ensure_product_queries(dict(product)) if product.get("details"): state["skip"] = state.get("skip", 0) + 1 products[i] = product continue url = None for key in family_candidates(product): if key in by_fam: url = by_fam[key] break if not url: try: url = discover_product_url(product.get("name") or "", market=args.market) except Exception: url = None if not url: state["fail"] = state.get("fail", 0) + 1 state.setdefault("errors", []).append( {"id": product.get("id"), "name": product.get("name"), "error": "no KIKO url"} ) products[i] = product enriched += 1 continue try: fetched = fetch_kiko_product(url=url, market=args.market) details = fetched.get("details") or [] if details: product["details"] = details if fetched.get("detailQueries") and not product.get("detailQueries"): product["detailQueries"] = fetched["detailQueries"] if fetched.get("sourceUrl"): product["sourceUrl"] = fetched["sourceUrl"] if fetched.get("sourceDescription") and not (product.get("description") or "").strip(): product["description"] = fetched.get("description") or fetched["sourceDescription"] local_vars = list(product.get("colorVariants") or []) remote_vars = list(fetched.get("colorVariants") or []) if local_vars and remote_vars: by_code = { str(v.get("colorCode") or "").strip(): v for v in remote_vars if v.get("colorCode") } by_shade = { str(v.get("shade") or "").strip().lower(): v for v in remote_vars if v.get("shade") } for lv in local_vars: rv = ( by_code.get(str(lv.get("colorCode") or "").strip()) or by_shade.get(str(lv.get("shade") or "").strip().lower()) ) if not rv: continue if rv.get("details") and not lv.get("details"): lv["details"] = rv["details"] if rv.get("sourceUrl"): lv["sourceUrl"] = rv["sourceUrl"] product["colorVariants"] = local_vars product = ensure_product_queries(product) state["ok"] = state.get("ok", 0) + 1 except Exception as exc: # noqa: BLE001 state["fail"] = state.get("fail", 0) + 1 state.setdefault("errors", []).append( { "id": product.get("id"), "name": product.get("name"), "url": url, "error": str(exc), } ) products[i] = product enriched += 1 if enriched % 5 == 0 or enriched == 1: print( f"[{enriched}/{limit}] ok={state['ok']} fail={state['fail']} " f"skip={state.get('skip', 0)} last={product.get('name')}" ) out = { **src, "metadata": { **(src.get("metadata") or {}), "note": "604 families with KIKO detail tables where matched", "withDetails": sum(1 for p in products if p.get("details")), "buildState": { "ok": state["ok"], "fail": state["fail"], "skip": state.get("skip", 0), }, }, "products": products, } save_json(OUT, out) save_json(STATE, {**state, "errors": (state.get("errors") or [])[-80:]}) time.sleep(max(0.0, args.sleep)) products = [ensure_product_queries(dict(p)) for p in products] with_details = sum(1 for p in products if p.get("details")) out = { **src, "format": "kikomiko-importer-v2", "metadata": { **(src.get("metadata") or {}), "note": "604 families with KIKO detail tables where matched", "productFamilies": len(products), "withDetails": with_details, "withDescription": sum( 1 for p in products if (p.get("description") or "").strip() ), "buildState": { "ok": state["ok"], "fail": state["fail"], "skip": state.get("skip", 0), }, }, "products": products, } save_json(OUT, out) save_json(STATE, {**state, "errors": (state.get("errors") or [])[-80:]}) print(f"Wrote {OUT}") print(f"withDetails={with_details}/{len(products)} ok={state['ok']} fail={state['fail']}") return 0 if __name__ == "__main__": raise SystemExit(main())