240 lines
8.4 KiB
Python
240 lines
8.4 KiB
Python
#!/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())
|