Ben_Kosytorz
4bbaefb45a
- warhammer_data.py: GW naming throughout - Space Marine chapters are
their own factions under a 'Space Marines' group, systems match the
shop (Age of Sigmar, The Horus Heresy, Other Games, The Old World,
Middle-Earth), classes are GW shop categories (Battleline, Infantry,
Vehicle, Walker, Flyer, Monster, Mounted, Hero, Titanic), group map
extended to AoS grand alliances / Other Games / Old World / Middle-Earth
- gwshop.py: hierarchy remap - GW depth-4 paths are system/group/faction/
class (the shop's last level is the class layer), depth-5 sources keep
a subfaction before it; catalog snapshot regenerated accordingly
- sorter.py: catalog entry is hierarchy-authoritative (group, class, and
faction when GW moved a unit or keyword tables are stale); generic GW
branches ('Unit Type') and a Battleline default never degrade a
confident keyword classification; whole-catalog retry when the faction
pool has no confident match (handles units GW moved between factions);
whole-catalog matching also runs when keyword classification found no
faction; invalid (HTML) items never fuzzy-match; default catalog is
now data/gwshop_catalog.json
- catalogue.py: --apply moves include the group layer, same structure
as the sorter
- docs/SORTING.md: hierarchy rewritten to GW naming with examples
- tests: expectations updated to GW naming (chapters, unit types)
- verified live: 4/4 real files in the test batch now catalog-match,
e.g. The Twin Lance -> 40k/Xenos Armies/T'au Empire/Epic Hero (GW's
current placement) instead of the stale Drukhari keyword
344 lines
12 KiB
Python
344 lines
12 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)
|
|
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, 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()
|