grokbot_ty
85786f16f2
- Taxonomy sync keeps fig-/feat- by slug (hyphen), not name; HTML-unescape names - Review: remove AI assist labels; Features picker for feat-*; derive fig_tag_ids from Figure term names (casefold match, no bridges); decodeName on display - Unit tests for scoped_tags, fig match, Review UI contract; docs slug wording
158 lines
6.6 KiB
Python
158 lines
6.6 KiB
Python
"""Taxonomy sync: pull Woo categories/tags/attributes/terms into the local cache.
|
|
|
|
Rebuild happens in one transaction; on any failure the prior cache is kept.
|
|
Only fig-* / feat-* tag families are retained by **slug** prefix with hyphen
|
|
(SETTINGS-51 / METADATA-15). taxonomy_anchor is a stable hash of the stored
|
|
subset, recorded only when it changes. Display names are HTML-unescaped so
|
|
Woo entities like ``&`` store as ``&``.
|
|
"""
|
|
|
|
import hashlib
|
|
import html
|
|
import json
|
|
from datetime import datetime, timezone
|
|
|
|
from .woo import WooClient
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(timezone.utc).replace(microsecond=0, tzinfo=None).isoformat()
|
|
|
|
|
|
def _decode_name(name: str | None) -> str:
|
|
"""Decode HTML entities Woo puts in taxonomy display names."""
|
|
return html.unescape(name or "")
|
|
|
|
|
|
def _scoped_tags(tags: list[dict]) -> list[dict]:
|
|
"""Keep product tags whose slug starts with fig- or feat- (hyphen required).
|
|
|
|
Live Woo uses human display names (e.g. "Curl") with slugs like fig-curl.
|
|
Filtering by name prefix kept 0 live tags; slug is authoritative.
|
|
"""
|
|
out = []
|
|
for t in tags:
|
|
slug = t.get("slug") or ""
|
|
if slug.startswith("fig-") or slug.startswith("feat-"):
|
|
out.append(t)
|
|
return out
|
|
|
|
|
|
def _normalize_category(c: dict) -> tuple:
|
|
"""Map a Woo REST category row into woo_categories insert values.
|
|
|
|
WooCommerce REST v3 returns `parent` (0 = top-level). Older fixtures may
|
|
already use `parent_id`. Never invent ids — only store what the payload carries.
|
|
"""
|
|
raw_parent = c.get("parent_id", c.get("parent"))
|
|
if raw_parent in (0, "0", None, ""):
|
|
parent_id = None
|
|
else:
|
|
parent_id = int(raw_parent)
|
|
woo_id = c.get("woo_id", c.get("id"))
|
|
if woo_id is not None:
|
|
woo_id = int(woo_id)
|
|
return (c["id"], _decode_name(c["name"]), c["slug"], parent_id, woo_id)
|
|
|
|
|
|
def _subset_hash(categories: list[dict], attributes: list[dict],
|
|
terms_by_attr: dict, scoped_tags: list[dict]) -> str:
|
|
subset = {
|
|
"categories": sorted(categories, key=lambda c: c["id"]),
|
|
"attributes": sorted(attributes, key=lambda a: a["id"]),
|
|
"terms": {str(k): sorted(v, key=lambda t: t["id"]) for k, v in terms_by_attr.items()},
|
|
"tags": sorted(scoped_tags, key=lambda t: t["id"]),
|
|
}
|
|
return hashlib.sha256(json.dumps(subset, sort_keys=True).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def sync(client: WooClient, conn) -> dict:
|
|
"""Rebuild the taxonomy cache from the store. Raises WooError on failure (caller
|
|
maps to 502; prior cache is retained because we only commit on success)."""
|
|
categories = client.get_categories()
|
|
tags = client.get_tags()
|
|
attributes = client.get_attributes()
|
|
terms_by_attr = {a["id"]: client.get_attribute_terms(a["id"]) for a in attributes}
|
|
scoped_tags = _scoped_tags(tags)
|
|
|
|
# Attributes/tags/terms are not referenced by slabs FKs — safe to rebuild.
|
|
conn.execute("DELETE FROM woo_attribute_terms")
|
|
conn.execute("DELETE FROM woo_attributes")
|
|
conn.execute("DELETE FROM woo_tags")
|
|
|
|
# Categories: upsert then drop stale rows. A full DELETE fails when slabs or
|
|
# pricing_rules still reference species_id (live UAT IntegrityError on test-woo).
|
|
cat_rows = [_normalize_category(c) for c in categories]
|
|
cat_ids = [r[0] for r in cat_rows]
|
|
for row in cat_rows:
|
|
conn.execute(
|
|
"INSERT INTO woo_categories(id, name, slug, parent_id, woo_id) VALUES (?,?,?,?,?) "
|
|
"ON CONFLICT(id) DO UPDATE SET "
|
|
"name=excluded.name, slug=excluded.slug, parent_id=excluded.parent_id, woo_id=excluded.woo_id",
|
|
row,
|
|
)
|
|
if cat_ids:
|
|
placeholders = ",".join("?" * len(cat_ids))
|
|
conn.execute(
|
|
f"DELETE FROM woo_categories WHERE id NOT IN ({placeholders}) "
|
|
"AND id NOT IN (SELECT species_id FROM slabs WHERE species_id IS NOT NULL) "
|
|
"AND id NOT IN (SELECT species_id FROM pricing_rules)",
|
|
cat_ids,
|
|
)
|
|
else:
|
|
conn.execute(
|
|
"DELETE FROM woo_categories "
|
|
"WHERE id NOT IN (SELECT species_id FROM slabs WHERE species_id IS NOT NULL) "
|
|
"AND id NOT IN (SELECT species_id FROM pricing_rules)"
|
|
)
|
|
|
|
conn.executemany(
|
|
"INSERT INTO woo_attributes(id, name, slug, woo_id) VALUES (?,?,?,?)",
|
|
[(a["id"], _decode_name(a["name"]), a["slug"], a.get("woo_id")) for a in attributes],
|
|
)
|
|
for attr_id, terms in terms_by_attr.items():
|
|
conn.executemany(
|
|
"INSERT INTO woo_attribute_terms(id, attribute_id, name, slug, woo_id) VALUES (?,?,?,?,?)",
|
|
[(t["id"], attr_id, _decode_name(t["name"]), t["slug"], t.get("woo_id")) for t in terms],
|
|
)
|
|
conn.executemany(
|
|
"INSERT INTO woo_tags(id, name, slug, woo_id) VALUES (?,?,?,?)",
|
|
[(t["id"], _decode_name(t["name"]), t["slug"], t.get("woo_id")) for t in scoped_tags],
|
|
)
|
|
|
|
anchor = _subset_hash(categories, attributes, terms_by_attr, scoped_tags)
|
|
now = _now_iso()
|
|
conn.execute("INSERT OR REPLACE INTO settings(key, value, value_enc) VALUES ('taxonomy_last_sync_at', ?, NULL)", (now,))
|
|
conn.execute("INSERT OR REPLACE INTO settings(key, value, value_enc) VALUES ('taxonomy_anchor', ?, NULL)", (anchor,))
|
|
conn.commit()
|
|
return {"taxonomy_anchor": anchor, "taxonomy_last_sync_at": now}
|
|
|
|
|
|
def get_cached(conn) -> dict:
|
|
"""Return the cached taxonomy grouped for the admin view."""
|
|
categories = [dict(r) for r in conn.execute("SELECT * FROM woo_categories ORDER BY name").fetchall()]
|
|
attributes = [dict(r) for r in conn.execute("SELECT * FROM woo_attributes ORDER BY name").fetchall()]
|
|
terms = [dict(r) for r in conn.execute("SELECT * FROM woo_attribute_terms ORDER BY name").fetchall()]
|
|
tags = [dict(r) for r in conn.execute("SELECT * FROM woo_tags ORDER BY name").fetchall()]
|
|
anchor_row = conn.execute("SELECT value FROM settings WHERE key='taxonomy_anchor'").fetchone()
|
|
sync_row = conn.execute("SELECT value FROM settings WHERE key='taxonomy_last_sync_at'").fetchone()
|
|
terms_by_attr: dict[int, list] = {a["id"]: [] for a in attributes}
|
|
for t in terms:
|
|
terms_by_attr.setdefault(t["attribute_id"], []).append(t)
|
|
grouped_attrs = [
|
|
{
|
|
"id": a["id"],
|
|
"name": a["name"],
|
|
"slug": a["slug"],
|
|
"terms": [{"id": t["id"], "name": t["name"], "slug": t["slug"]} for t in terms_by_attr.get(a["id"], [])],
|
|
}
|
|
for a in attributes
|
|
]
|
|
return {
|
|
"categories": categories,
|
|
"attributes": grouped_attrs,
|
|
"tags": tags,
|
|
"taxonomy_anchor": anchor_row[0] if anchor_row else None,
|
|
"taxonomy_last_sync_at": sync_row[0] if sync_row else None,
|
|
}
|