Ty
d83c9d5c21
- backend/woo.py: stdlib urllib Woo v3 client. Basic-auth, HTTPS-only, X-WP-TotalPages pagination, typed errors (CredentialsMissing/Auth/NotFound/Server/Connection), injectable transport for tests, WP core media upload. - backend/taxonomy.py: sync pulls categories/tags/attributes/terms into cache in one commit (rollback keeps prior cache on failure); fig-*/feat-* tag scoping; stable taxonomy_anchor hash; get_cached grouped admin view. - backend/app.py: POST /settings/test-woo (fail-closed, resync on success), GET /admin/taxonomy, POST /admin/taxonomy/sync (502 on Woo failure). - frontend: Settings Test Woo button + result display. - tests: 8 woo/taxonomy tests (auth header, non-https refusal, pagination, missing creds, 401, scoped sync, failure-keeps-cache, anchor stability). Verified: pytest 31 passed; check_requirements + check_coverage exit 0; npm build ok; live QA test-woo fail-closed + sync 502.
100 lines
4.4 KiB
Python
100 lines
4.4 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 (METADATA-15). taxonomy_anchor is a
|
|
stable hash of the stored subset, recorded only when it changes.
|
|
"""
|
|
|
|
import hashlib
|
|
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 _scoped_tags(tags: list[dict]) -> list[dict]:
|
|
return [t for t in tags if t.get("name", "").startswith(("fig-", "feat-"))]
|
|
|
|
|
|
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)
|
|
|
|
conn.execute("DELETE FROM woo_attribute_terms")
|
|
conn.execute("DELETE FROM woo_attributes")
|
|
conn.execute("DELETE FROM woo_categories")
|
|
conn.execute("DELETE FROM woo_tags")
|
|
conn.executemany(
|
|
"INSERT INTO woo_categories(id, name, slug, parent_id, woo_id) VALUES (?,?,?,?,?)",
|
|
[(c["id"], c["name"], c["slug"], c.get("parent_id"), c.get("woo_id")) for c in categories],
|
|
)
|
|
conn.executemany(
|
|
"INSERT INTO woo_attributes(id, name, slug, woo_id) VALUES (?,?,?,?)",
|
|
[(a["id"], 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, 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"], 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,
|
|
}
|