fix(m3): Option A must-fix #79 #82 #83 #84 #89 #98

- #79 publish SKU/create via WooClient (User-Agent + typed 403 unrecoverable)
- #82 stale_field current_options from live taxonomy + owner labels
- #83 create allowlist/extra-key safe; taxonomy+enum 4xx on create
- #84 health woo_reachable real probe (non-failing); version 0.5.0-m3
- #89 review persists assembled title/short after taxonomy lock; UI applies
- #98 live-edge from synced term; attrs resolved by slug then name
This commit is contained in:
Ty
2026-09-21 12:49:53 -07:00
parent 96a697a50a
commit 0c00d25cc7
7 changed files with 328 additions and 108 deletions
+38 -13
View File
@@ -98,14 +98,7 @@ class WooClient:
def _get(self, path: str) -> tuple[int, dict, bytes]:
url = f"{self.base_url}/wp-json/wc/v3{path}"
return self._request(
url, "GET", None,
{
"Authorization": f"Basic {self._auth}",
"Accept": "application/json",
"User-Agent": _USER_AGENT,
},
)
return self._request(url, "GET", None, self._auth_headers())
def _fetch_all(self, path: str) -> list[dict[str, Any]]:
page = 1
@@ -132,6 +125,42 @@ class WooClient:
def get_attribute_terms(self, attr_id: int) -> list[dict[str, Any]]:
return self._fetch_all(f"/products/attributes/{attr_id}/terms")
def _auth_headers(self, extra: dict[str, str] | None = None) -> dict[str, str]:
"""Shared headers for every outbound Woo/WP request (User-Agent + auth)."""
h = {
"Authorization": f"Basic {self._auth}",
"Accept": "application/json",
"User-Agent": _USER_AGENT,
}
if extra:
h.update(extra)
return h
def get_products_by_sku(self, sku: str) -> list[dict[str, Any]]:
"""GET products filtered by SKU. Raises typed Woo* errors on non-200."""
code, _h, body = self._request(
f"{self.base_url}/wp-json/wc/v3/products?sku={sku}",
"GET",
None,
self._auth_headers(),
)
if code != 200:
raise WooServerError(f"sku lookup returned {code}")
return list(json.loads(body))
def create_product(self, payload: dict[str, Any]) -> int:
"""POST /products. Raises typed Woo* errors on failure; returns product id."""
body = json.dumps(payload).encode("utf-8")
code, _h, resp = self._request(
f"{self.base_url}/wp-json/wc/v3/products",
"POST",
body,
self._auth_headers({"Content-Type": "application/json"}),
)
if code not in (200, 201):
raise WooServerError(f"product create returned {code}")
return int(json.loads(resp)["id"])
def upload_media(self, png_bytes: bytes, filename: str) -> int:
"""POST a PNG to WordPress core media; return the attachment id."""
boundary = "----slabuploader-media-boundary"
@@ -143,11 +172,7 @@ class WooClient:
url = f"{self.base_url}/wp-json/wp/v2/media"
code, _headers, resp = self._request(
url, "POST", body,
{
"Authorization": f"Basic {self._auth}",
"Content-Type": f"multipart/form-data; boundary={boundary}",
"User-Agent": _USER_AGENT,
},
self._auth_headers({"Content-Type": f"multipart/form-data; boundary={boundary}"}),
)
if code not in (200, 201):
raise WooServerError(f"media upload returned {code}")