"""WooCommerce REST v3 client (httpx, WordPress app-password basic auth, HTTPS-only). Creds come from Settings (wp_username plain, app_password AES-GCM). Missing creds raise WooCredentialsMissing (a fail-closed condition the caller maps to ok:false), distinct from WooAuthError (a 401 from the store). The transport is a single injectable call so tests never hit a real store (mock boundary). """ from __future__ import annotations import base64 import json from typing import Any, Callable import httpx # WordPress security plugins (Wordfence, WAF, CDN) commonly block the default # Python User-Agent with 403. Send a descriptive one (#79). _USER_AGENT = "SlabUploader/1.0 (+https://www.whidbeywoodstore.com)" class WooError(Exception): """Base for all Woo client failures.""" class WooCredentialsMissing(WooError): """Woo base URL or credentials are not configured (fail-closed).""" class WooAuthError(WooError): """The store rejected the credentials (401).""" class WooForbiddenError(WooError): """The store blocked the request (403) — security plugin / WAF.""" class WooNotFoundError(WooError): """The store returned 404.""" class WooServerError(WooError): """The store returned a 5xx or an unexpected status.""" class WooConnectionError(WooError): """Store unreachable (DNS, TLS, timeout).""" Transport = Callable[[str, str, bytes | None, dict[str, str]], tuple[int, dict, bytes]] def _default_request(url: str, method: str, data: bytes | None, headers: dict[str, str]) -> tuple[int, dict, bytes]: try: with httpx.Client(timeout=15.0) as client: resp = client.request(method, url, content=data, headers=headers) except (httpx.TimeoutException, httpx.TransportError) as e: raise WooConnectionError(f"store unreachable: {e}") from e return resp.status_code, dict(resp.headers.items()), resp.content def _message(body: bytes) -> str | None: try: parsed = json.loads(body) except (ValueError, UnicodeDecodeError): return None if isinstance(parsed, dict): msg = parsed.get("message") if isinstance(msg, str) and msg: return msg return None class WooClient: """Minimal Woo REST v3 client. `request` is the transport seam for tests.""" def __init__(self, username: str, password: str, base_url: str, request: Transport | None = None) -> None: if not base_url or not base_url.startswith("https://"): raise WooCredentialsMissing("WOO_BASE_URL must be https:// (SETTINGS-10)") self.base_url = base_url.rstrip("/") self._request: Transport = request or _default_request self._auth = base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii") @classmethod def from_settings(cls, base_url: str, username: str | None, encrypted_password: str | None, request: Transport | None = None) -> "WooClient": """Build from Settings values; raise WooCredentialsMissing if incomplete.""" from .crypto import DecryptError, decrypt if not username: raise WooCredentialsMissing("woo_wp_username not set (SETTINGS-15)") if not encrypted_password: raise WooCredentialsMissing("woo_app_password not configured (SETTINGS-15)") try: password = decrypt(encrypted_password) except DecryptError: raise WooCredentialsMissing("woo_app_password cannot be decrypted (re-enter it)") from None return cls(username, password, base_url, request) def _auth_headers(self, extra: dict[str, str] | None = None) -> dict[str, str]: h = { "Authorization": f"Basic {self._auth}", "Accept": "application/json", "User-Agent": _USER_AGENT, } if extra: h.update(extra) return h def _raise_for(self, code: int, body: bytes) -> None: if code == 401: raise WooAuthError(_message(body) or "authentication rejected (401)") if code == 403: raise WooForbiddenError(_message(body) or "request forbidden (403)") if code == 404: raise WooNotFoundError(_message(body) or "not found (404)") raise WooServerError(f"server error {code}: {_message(body) or ''}".strip(": ")) def _get(self, path: str) -> tuple[int, dict, bytes]: return self._request(f"{self.base_url}/wp-json/wc/v3{path}", "GET", None, self._auth_headers()) def _fetch_all(self, path: str) -> list[dict[str, Any]]: page = 1 results: list[dict[str, Any]] = [] while True: code, headers, body = self._get(f"{path}?per_page=100&page={page}") if code != 200: self._raise_for(code, body) results.extend(json.loads(body)) total_pages = int(headers.get("X-WP-TotalPages") or headers.get("x-wp-totalpages") or 1) if page >= total_pages: return results page += 1 def get_categories(self) -> list[dict[str, Any]]: return self._fetch_all("/products/categories") def get_tags(self) -> list[dict[str, Any]]: return self._fetch_all("/products/tags") def get_attributes(self) -> list[dict[str, Any]]: return self._fetch_all("/products/attributes") def get_attribute_terms(self, attr_id: int) -> list[dict[str, Any]]: return self._fetch_all(f"/products/attributes/{attr_id}/terms") def get_products_by_sku(self, sku: str) -> list[dict[str, Any]]: code, _h, body = self._request( f"{self.base_url}/wp-json/wc/v3/products?sku={sku}", "GET", None, self._auth_headers() ) if code != 200: self._raise_for(code, body) return list(json.loads(body)) def create_product(self, payload: dict[str, Any]) -> int: code, _h, resp = self._request( f"{self.base_url}/wp-json/wc/v3/products", "POST", json.dumps(payload).encode("utf-8"), self._auth_headers({"Content-Type": "application/json"}), ) if code not in (200, 201): self._raise_for(code, resp) return int(json.loads(resp)["id"]) def upload_media(self, png_bytes: bytes, filename: str) -> int: """POST a PNG to WordPress core media; returns the attachment id.""" boundary = "----slabuploader-media-boundary" body = ( f"--{boundary}\r\n" f'Content-Disposition: form-data; name="file"; filename="{filename}"\r\n' "Content-Type: image/png\r\n\r\n" ).encode("utf-8") + png_bytes + f"\r\n--{boundary}--\r\n".encode("utf-8") code, _h, resp = self._request( f"{self.base_url}/wp-json/wp/v2/media", "POST", body, self._auth_headers({"Content-Type": f"multipart/form-data; boundary={boundary}"}), ) if code not in (200, 201): self._raise_for(code, resp) return int(json.loads(resp)["id"])