Add GW webshop catalog source (Algolia): dlphelper.gwshop + full dump

- warhammer.com shop HTML/listing endpoints are behind AWS WAF (405/202
  goku challenge) for non-browser clients; findings recorded in AGENTS.md
- the shop search runs on Algolia; public search-only key lets us dump the
  whole product index with plain requests (no browser, no captcha)
- gwshop.py: credential loading (env or gitignored algolia.json), index
  listing, facet-partitioned full retrieval (search-only key cannot /browse
  and pagination caps at 1000, so the index is partitioned on the
  GameSystemsRoot hierarchy facet), normalization into the sorter's
  source-agnostic catalog schema (system/group/faction/subfaction/class)
- data/gwshop_catalog.json: snapshot of prod-lazarus-product-en-ww,
  3491 unique products, every product with full hierarchy path
- tools/: one-shot browse + catalog builder + shape inspector

.gitignore

@@ -18,3 +18,7 @@ manifest.jsonl
 # Never commit credentials
 *.token
 .env
+
+# Algolia credentials (search-only key, still never commit)
+algolia.json
+gwshop_raw.json

AGENTS.md

@@ -199,7 +199,24 @@ never re-fetch them again):
    `catalogue_moves.jsonl`). The taxonomy lives in
    `dlphelper/warhammer_data.py`; it is keyword-based and grows with the
    collection - unclassified files are reported, never guessed. Manual fixes
-   go into an overrides JSON (`--overrides`).
+   goes into an overrides JSON (`--overrides`).
+
+   **Catalog source findings (warhammer.com, verified 2026-10-04, do not
+   re-probe blindly):** the shop HTML and all search/listing endpoints
+   (`/en-WW/shop/*`, `/app/resources/catalog/search/*`,
+   `/app/resources/catalog/categories`) sit behind an **AWS WAF** challenge
+   (405/202 "goku" JS captcha) for any non-browser client, regardless of
+   User-Agent. Static assets, robots.txt and 404 pages are served normally.
+   The sitemap declared in robots.txt does not exist. Deep paths under the
+   allowed `/app/resources/catalog/product/*` reach a separate commerce
+   backend (plain `404 Not found`) but the correct resource shape is
+   unknown. The site's product search is **Algolia**; its app-id and
+   search-only key are public (they ship with the browser page) and would
+   allow a full catalog dump via plain `requests` — this is the most
+   promising path: capture one search request from the browser devtools
+   ("Copy as cURL") and build `dlphelper/gwshop.py` on it. Catalog sources
+   stay pluggable and source-agnostic regardless (miniset.net, manual JSON
+   overrides, GW shop).
 4. **Watch mode**: `--subs subscriptions.json` + a cron/timer already gives a
    primitive version; a real watch mode (per-subscription schedules, tear-off
    notifications, discord pings on new links) builds on it.

dlphelper/gwshop.py

@@ -0,0 +1,334 @@
+"""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)
+        return self._partition(index, [], total)
+
+    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] = []
+        matched = 0
+        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)
+                matched += nb
+                log.info(
+                    "gwshop %s: %s -> %d hits", index, sub_filters[:80], nb
+                )
+            else:
+                sub = self._partition(index, sub_path, count)
+                out.extend(sub)
+                matched += len(sub)
+        residual = total - matched
+        if residual > 0:
+            log.warning(
+                "gwshop %s: %d hits do not carry %s.lvl%d "
+                "(unclassified products, 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, Armies of the Imperium, Adeptus Custodes, Rules & Accessories]
+        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
+        subfaction = primary[3] if len(primary) > 3 else None
+        # 5th level (if present) is treated as class; GW uses "Rules &
+        # Accessories" etc. rather than force-org classes.
+        klass = primary[4] if len(primary) > 4 else 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()

tools/gw_browse_test.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python3
+"""One-shot browse of the GW Algolia index -> gwshop_raw.json (gitignored)."""
+import json
+import logging
+import pathlib
+import sys
+
+sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
+
+import dlphelper.gwshop as gw
+
+logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+
+app, key = gw.load_credentials()
+client = gw.GwShopClient(app, key)
+hits = client.browse("prod-lazarus-product-en-ww", hits_per_page=100)
+print("total hits", len(hits))
+pathlib.Path("gwshop_raw.json").write_text(
+    json.dumps(hits, ensure_ascii=False), encoding="utf-8"
+)
+print("written gwshop_raw.json")

tools/gw_build_catalog.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python3
+"""gwshop_raw.json -> data/gwshop_catalog.json + quick shape report."""
+import json
+import logging
+import pathlib
+import sys
+from collections import Counter
+
+sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
+
+import dlphelper.gwshop as gw
+
+logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+
+hits = json.loads(pathlib.Path("gwshop_raw.json").read_text(encoding="utf-8"))
+print("raw hits:", len(hits))
+unique = len({h.get("objectID") for h in hits})
+print("unique objectIDs:", unique)
+
+catalog = gw.build_catalog(hits)
+pathlib.Path("data/gwshop_catalog.json").write_text(
+    json.dumps(catalog, ensure_ascii=False, indent=1), encoding="utf-8"
+)
+print("entries:", catalog["generated_entries"])
+
+entries = catalog["entries"]
+print("no system:", sum(1 for e in entries if e["system"] is None))
+print("by system:", Counter(e["system"] for e in entries).most_common())
+print("40k groups:", Counter(
+    e["group"] for e in entries if e["system"] == "Warhammer 40,000"
+).most_common())
+print("lvl depth:", Counter(
+    len((e.get("gw_paths") or [[]])[0]) for e in entries
+).most_common())
+print("classes (lvl>=4):", Counter(
+    e["class"] for e in entries if e["class"]
+).most_common(12))
+sample = next(
+    e for e in entries
+    if e["name"].startswith("Codex: Adeptus Custodes")
+)
+print(json.dumps(sample, indent=1, ensure_ascii=False)[:900])

tools/gw_inspect.py

@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+"""Inspect gwshop catalog shape: Space Marines chapters, AoS, dedupe stats."""
+import json
+import pathlib
+from collections import Counter
+
+cat = json.loads(pathlib.Path("data/gwshop_catalog.json").read_text(encoding="utf-8"))
+entries = cat["entries"]
+
+def show(pred, label, n=6):
+    print(f"--- {label}")
+    shown = 0
+    for e in entries:
+        if pred(e):
+            p = " > ".join(x for x in (e["system"], e["group"], e["faction"], e["subfaction"], e["class"]) if x)
+            print(f"  {e['name'][:60]!r}  ->  {p}")
+            shown += 1
+            if shown >= n:
+                break
+
+show(lambda e: e["group"] == "Space Marines", "Space Marines group samples")
+show(lambda e: e["group"] == "Armies of the Imperium" and e["subfaction"], "Imperium with subfaction")
+show(lambda e: e["system"] == "Age of Sigmar", "Age of Sigmar samples", 5)
+show(lambda e: e["system"] == "The Old World", "Old World samples", 5)
+show(lambda e: e["system"] == "Other Games", "Other Games samples", 5)
+show(lambda e: e["group"] == "Unit Type", "Unit Type branch", 6)
+
+names = Counter(e["name"].casefold() for e in entries)
+dups = {n: c for n, c in names.items() if c > 1}
+print("duplicate names:", len(dups), "example:", list(dups.items())[:3])
+
+# faction-level value census for 40k
+print("--- 40k factions census")
+print(Counter(e["faction"] for e in entries if e["system"] == "Warhammer 40,000").most_common(40))