Ben_Kosytorz
7e28518a52
Summed nbHits across facet branches double-counted products listed in several branches; coverage is now per unique objectID, with a warning about records left out of the dump. AGENTS.md documents the live count (4183 records, 3491 via facet).
354 lines
13 KiB
Python
354 lines
13 KiB
Python
"""Games Workshop webshop catalog via Algolia.
|
|
|
|
The warhammer.com shop is behind an AWS WAF challenge for non-browser
|
|
clients, but its product search runs on Algolia and the app-id plus the
|
|
public search-only key ship with every browser page. This module uses them
|
|
to enumerate the whole product index and normalize it into the sorter's
|
|
catalog schema (system/group/faction/subfaction/class/name/aliases).
|
|
|
|
Credentials are read from (first hit wins):
|
|
1. env ALGOLIA_APP_ID + ALGOLIA_API_KEY
|
|
2. algolia.json next to the repo root (gitignored):
|
|
{"app_id": "...", "api_key": "..."}
|
|
|
|
The key is the *search-only* public key (never an admin key). Treat any
|
|
key found in algolia.json as still secret: never log or commit it.
|
|
|
|
CLI: python3 -m dlphelper.gwshop OUT.json [--index NAME] [--raw RAW.json]
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
DEFAULT_INDEX = "prod-lazarus-product-en-ww"
|
|
# The read host (-dsn) serves searches; credentials also work as query
|
|
# parameters, which keeps every request a plain GET/POST with no headers
|
|
# beyond content-type.
|
|
_DSN_HOST = "https://{app_id}-dsn.algolia.net"
|
|
|
|
_PAGE_SIZE = 1000 # Algolia max hitsPerPage per /query page
|
|
_RETRY_DELAYS = (2, 5, 15)
|
|
|
|
|
|
def load_credentials() -> tuple[str, str]:
|
|
"""Resolve (app_id, api_key) from env or algolia.json."""
|
|
app_id = os.environ.get("ALGOLIA_APP_ID", "")
|
|
api_key = os.environ.get("ALGOLIA_API_KEY", "")
|
|
if app_id and api_key:
|
|
return app_id, api_key
|
|
cfg = Path(__file__).resolve().parent.parent / "algolia.json"
|
|
if cfg.is_file():
|
|
data = json.loads(cfg.read_text(encoding="utf-8"))
|
|
app_id = data.get("app_id", "") or app_id
|
|
api_key = data.get("api_key", "") or api_key
|
|
if not app_id or not api_key:
|
|
raise SystemExit(
|
|
"Algolia credentials missing: set ALGOLIA_APP_ID and "
|
|
"ALGOLIA_API_KEY, or create algolia.json with "
|
|
'{"app_id": "...", "api_key": "..."} (gitignored).'
|
|
)
|
|
return app_id, api_key
|
|
|
|
|
|
class GwShopClient:
|
|
"""Minimal Algolia search client for one application."""
|
|
|
|
def __init__(self, app_id: str, api_key: str) -> None:
|
|
self.base = _DSN_HOST.format(app_id=app_id)
|
|
self.params = {
|
|
"x-algolia-application-id": app_id,
|
|
"x-algolia-api-key": api_key,
|
|
}
|
|
self.session = requests.Session()
|
|
|
|
def _request(self, method: str, path: str, body: dict | None = None) -> Any:
|
|
url = f"{self.base}{path}"
|
|
last_err: Exception | None = None
|
|
for attempt, delay in enumerate((0,) + _RETRY_DELAYS, start=1):
|
|
if delay:
|
|
log.info("Algolia retry %d in %ds", attempt, delay)
|
|
time.sleep(delay)
|
|
try:
|
|
r = self.session.request(
|
|
method,
|
|
url,
|
|
params=self.params,
|
|
json=body,
|
|
timeout=30,
|
|
)
|
|
if r.status_code == 200:
|
|
return r.json()
|
|
last_err = RuntimeError(
|
|
f"Algolia {method} {path} -> {r.status_code}: {r.text[:200]}"
|
|
)
|
|
if r.status_code in (401, 403): # credentials: no point retrying
|
|
break
|
|
except requests.RequestException as exc:
|
|
last_err = exc
|
|
raise RuntimeError(f"Algolia request failed: {last_err}") from last_err
|
|
|
|
def list_indices(self) -> list[dict]:
|
|
return self._request("GET", "/1/indexes").get("items", [])
|
|
|
|
# ---- full index retrieval -------------------------------------------
|
|
# The search-only key cannot use /browse (403) and normal pagination is
|
|
# capped at 1000 cumulative results. So we partition the index on the
|
|
# GameSystemsRoot hierarchical facet: each branch is fetched with an
|
|
# explicit filter, recursing deeper only when a branch exceeds one page.
|
|
|
|
_FACET = "GameSystemsRoot"
|
|
|
|
@staticmethod
|
|
def _escape(value: str) -> str:
|
|
return value.replace("\\", "\\\\").replace('"', '\\"')
|
|
|
|
def _filter_expr(self, path: list[tuple[int, str]]) -> str:
|
|
return " AND ".join(
|
|
f'{self._FACET}.lvl{lvl}:"{self._escape(value)}"'
|
|
for lvl, value in path
|
|
)
|
|
|
|
def _search_all(self, index: str, filters: str | None) -> tuple[list[dict], int]:
|
|
"""All hits matching `filters` (must be <= pagination cap)."""
|
|
out: list[dict] = []
|
|
params = f"query=&hitsPerPage={_PAGE_SIZE}&page=0"
|
|
if filters:
|
|
params += f"&filters={requests.utils.quote(filters, safe='')}"
|
|
page = self._request(
|
|
"POST", f"/1/indexes/{index}/query", body={"params": params}
|
|
)
|
|
out.extend(page.get("hits", []))
|
|
return out, int(page.get("nbHits") or 0)
|
|
|
|
def _facet_counts(
|
|
self, index: str, level: int, filters: str | None
|
|
) -> dict[str, int]:
|
|
params = f"query=&hitsPerPage=0&facets={self._FACET}.lvl{level}"
|
|
if filters:
|
|
params += f"&filters={requests.utils.quote(filters, safe='')}"
|
|
page = self._request(
|
|
"POST", f"/1/indexes/{index}/query", body={"params": params}
|
|
)
|
|
facets = page.get("facets") or {}
|
|
return facets.get(f"{self._FACET}.lvl{level}") or {}
|
|
|
|
def browse(self, index: str, hits_per_page: int = _PAGE_SIZE) -> list[dict]:
|
|
"""Every record of the index, via facet-partitioned search."""
|
|
hits, total = self._search_all(index, None)
|
|
if total <= hits_per_page:
|
|
return hits
|
|
log.info("gwshop %s: %d hits, partitioning on %s", index, total, self._FACET)
|
|
out = self._partition(index, [], total)
|
|
covered = len({h.get("objectID") for h in out})
|
|
if covered < total:
|
|
log.warning(
|
|
"gwshop %s: %d of %d index records left out of the dump "
|
|
"(no fetchable %s branch path)",
|
|
index, total - covered, total, self._FACET,
|
|
)
|
|
return out
|
|
|
|
def _partition(
|
|
self, index: str, path: list[tuple[int, str]], total: int
|
|
) -> list[dict]:
|
|
level = len(path)
|
|
filters = self._filter_expr(path) if path else None
|
|
counts = self._facet_counts(index, level, filters)
|
|
out: list[dict] = []
|
|
covered: set[Any] = set()
|
|
for value, count in sorted(counts.items()):
|
|
sub_path = path + [(level, value)]
|
|
sub_filters = self._filter_expr(sub_path)
|
|
if count <= _PAGE_SIZE:
|
|
hits, nb = self._search_all(index, sub_filters)
|
|
out.extend(hits)
|
|
covered.update(h.get("objectID") for h in hits)
|
|
log.info(
|
|
"gwshop %s: %s -> %d hits", index, sub_filters[:80], nb
|
|
)
|
|
else:
|
|
sub = self._partition(index, sub_path, count)
|
|
out.extend(sub)
|
|
covered.update(h.get("objectID") for h in sub)
|
|
# nbHits summed across branches double-counts products listed in
|
|
# several branches, so coverage is counted per unique objectID.
|
|
residual = total - len(covered)
|
|
if residual > 0:
|
|
log.warning(
|
|
"gwshop %s: %d hits in branch carry no %s.lvl%d "
|
|
"(shallow facet path, left out of the dump)",
|
|
index, residual, self._FACET, level,
|
|
)
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# normalization
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _clean(node: str) -> str:
|
|
return node.strip()
|
|
|
|
|
|
def hierarchy_paths(hit: dict) -> list[list[str]]:
|
|
"""Every full path of the GameSystemsRoot hierarchical facet.
|
|
|
|
A hit may belong to several branches (e.g. an army box also being a
|
|
"Gaming Rules" entry); each branch is returned as its own path
|
|
[system, group, faction, subfaction_or_class, ...].
|
|
"""
|
|
root = hit.get("GameSystemsRoot") or {}
|
|
paths: list[list[str]] = []
|
|
for lvl_key in sorted(root, key=lambda k: int(k[3:])):
|
|
for value in root.get(lvl_key) or []:
|
|
paths.append([_clean(p) for p in value.split(" > ")])
|
|
return paths
|
|
|
|
|
|
def normalize(hit: dict) -> dict:
|
|
"""One Algolia hit -> one or more catalog entries (usually one)."""
|
|
name = _clean(hit.get("name") or hit.get("objectID") or "")
|
|
entry: dict[str, Any] = {
|
|
"name": name,
|
|
"aliases": [],
|
|
"sku": hit.get("sku"),
|
|
"slug": hit.get("slug"),
|
|
"productType": hit.get("productType"),
|
|
"source": "gwshop",
|
|
"gw_paths": hierarchy_paths(hit),
|
|
"collections": hit.get("_collections") or [],
|
|
}
|
|
return entry
|
|
|
|
|
|
def build_catalog(hits: list[dict]) -> dict:
|
|
"""Index dump -> catalog document in the sorter's schema shape.
|
|
|
|
The hierarchy layer (system/group/faction/subfaction) is expanded from
|
|
gw_paths; name/aliases stay per product so the fuzzy matcher works on
|
|
the product name, exactly like data/library_catalog.json entries.
|
|
"""
|
|
entries: list[dict] = []
|
|
seen: set[tuple] = set()
|
|
for hit in hits:
|
|
base = normalize(hit)
|
|
paths = base["gw_paths"]
|
|
if not paths:
|
|
key = (hit.get("objectID"), ())
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
entries.append(
|
|
{
|
|
"system": None,
|
|
"group": None,
|
|
"faction": None,
|
|
"subfaction": None,
|
|
"class": None,
|
|
"name": base["name"],
|
|
"aliases": [],
|
|
"sku": base["sku"],
|
|
"slug": base["slug"],
|
|
"productType": base["productType"],
|
|
"source": "gwshop",
|
|
"collections": base["collections"],
|
|
}
|
|
)
|
|
continue
|
|
# primary branch = longest path (most specific), e.g.
|
|
# [40k, Space Marines, Raven Guard, Walker]
|
|
primary = max(paths, key=len)
|
|
system = primary[0] if len(primary) > 0 else None
|
|
group = primary[1] if len(primary) > 1 else None
|
|
faction = primary[2] if len(primary) > 2 else None
|
|
# The LAST level is the class/category layer (GW uses shop
|
|
# categories there: Battleline, Walker, Vehicle, "Rules &
|
|
# Accessories", product lines like "Cadian"). Deeper sources
|
|
# (miniset style) may carry a real subfaction before it.
|
|
if len(primary) >= 5:
|
|
subfaction = primary[3]
|
|
klass = primary[4]
|
|
elif len(primary) == 4:
|
|
subfaction = None
|
|
klass = primary[3]
|
|
else:
|
|
subfaction = None
|
|
klass = None
|
|
key = (hit.get("objectID"), tuple(primary))
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
entries.append(
|
|
{
|
|
"system": system,
|
|
"group": group,
|
|
"faction": faction,
|
|
"subfaction": subfaction,
|
|
"class": klass,
|
|
"name": base["name"],
|
|
"aliases": [],
|
|
"sku": base["sku"],
|
|
"slug": base["slug"],
|
|
"productType": base["productType"],
|
|
"source": "gwshop",
|
|
"collections": base["collections"],
|
|
}
|
|
)
|
|
return {
|
|
"comment": (
|
|
"Generated by dlphelper.gwshop from the Games Workshop webshop "
|
|
"(Algolia index, deduplicated by product). Source-agnostic "
|
|
"catalog schema: system/group/faction/subfaction/class/name/aliases."
|
|
),
|
|
"source": "gwshop",
|
|
"generated_entries": len(entries),
|
|
"entries": entries,
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
|
|
parser = argparse.ArgumentParser(
|
|
description="Dump the GW webshop catalog via Algolia."
|
|
)
|
|
parser.add_argument("out", help="output catalog JSON path")
|
|
parser.add_argument("--index", default=DEFAULT_INDEX)
|
|
parser.add_argument("--raw", help="also write the raw Algolia hits here")
|
|
args = parser.parse_args()
|
|
|
|
app_id, api_key = load_credentials()
|
|
client = GwShopClient(app_id, api_key)
|
|
hits = client.browse(args.index)
|
|
log.info("%d raw hits from %s", len(hits), args.index)
|
|
|
|
if args.raw:
|
|
Path(args.raw).write_text(
|
|
json.dumps(hits, ensure_ascii=False, indent=1), encoding="utf-8"
|
|
)
|
|
log.info("raw hits written to %s", args.raw)
|
|
|
|
catalog = build_catalog(hits)
|
|
Path(args.out).write_text(
|
|
json.dumps(catalog, ensure_ascii=False, indent=1), encoding="utf-8"
|
|
)
|
|
log.info(
|
|
"%d catalog entries written to %s",
|
|
catalog["generated_entries"],
|
|
args.out,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|