Split 49 modules/suites into independent git repos; untrack from monorepo
Each top-level module/suite folder is now its own private repo on GitHub (gsinghpal/<name>) and gitea (admin/<name>), with a fresh single initial commit. The monorepo no longer tracks them (added to .gitignore + git rm --cached); working-tree files are retained on disk and managed in their own repos. The monorepo keeps shared root files (CLAUDE.md, docs/, scripts/, tools/, AGENTS.md, WIP/obsolete dirs) and full history. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsinghpal
@@ -1,3 +0,0 @@
|
||||
from .woo_api_client import WooApiClient
|
||||
from .ai_service import AIService
|
||||
from .image_processor import ImageProcessor
|
||||
@@ -1,171 +0,0 @@
|
||||
import json
|
||||
import logging
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class AIService:
|
||||
"""AI content generation service supporting Claude and OpenAI."""
|
||||
|
||||
def __init__(self, provider, api_key, model=None):
|
||||
"""
|
||||
Args:
|
||||
provider: 'claude' or 'openai'
|
||||
api_key: API key for the chosen provider
|
||||
model: Model name (defaults to claude-sonnet-4-5-20250514 or gpt-4o)
|
||||
"""
|
||||
self.provider = provider
|
||||
self.api_key = api_key
|
||||
if model:
|
||||
self.model = model
|
||||
else:
|
||||
self.model = 'claude-sonnet-4-5-20250514' if provider == 'claude' else 'gpt-4o'
|
||||
self._client = None
|
||||
|
||||
def _get_client(self):
|
||||
if self._client:
|
||||
return self._client
|
||||
if self.provider == 'claude':
|
||||
try:
|
||||
import anthropic
|
||||
self._client = anthropic.Anthropic(api_key=self.api_key)
|
||||
except ImportError:
|
||||
raise RuntimeError("anthropic package not installed. Run: pip install anthropic")
|
||||
elif self.provider == 'openai':
|
||||
try:
|
||||
import openai
|
||||
self._client = openai.OpenAI(api_key=self.api_key)
|
||||
except ImportError:
|
||||
raise RuntimeError("openai package not installed. Run: pip install openai")
|
||||
return self._client
|
||||
|
||||
def generate(self, system_prompt, user_message, max_tokens=2000):
|
||||
"""Generate text using the configured AI provider."""
|
||||
client = self._get_client()
|
||||
try:
|
||||
if self.provider == 'claude':
|
||||
response = client.messages.create(
|
||||
model=self.model,
|
||||
max_tokens=max_tokens,
|
||||
system=system_prompt,
|
||||
messages=[{"role": "user", "content": user_message}],
|
||||
)
|
||||
return response.content[0].text
|
||||
elif self.provider == 'openai':
|
||||
response = client.chat.completions.create(
|
||||
model=self.model,
|
||||
max_tokens=max_tokens,
|
||||
messages=[
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": user_message},
|
||||
],
|
||||
)
|
||||
return response.choices[0].message.content
|
||||
except Exception as e:
|
||||
_logger.error("AI generation failed (%s): %s", self.provider, str(e))
|
||||
raise
|
||||
|
||||
def generate_product_content(self, product_info, prompts):
|
||||
"""Generate all product content at once.
|
||||
|
||||
Args:
|
||||
product_info: dict with keys like name, category, features, raw_description
|
||||
prompts: dict with keys: title, short_desc, long_desc, meta_title, meta_desc, keywords
|
||||
|
||||
Returns:
|
||||
dict with generated content for each field
|
||||
"""
|
||||
context = json.dumps(product_info, indent=2)
|
||||
system = (
|
||||
"You are an expert e-commerce copywriter and SEO specialist. "
|
||||
"You create compelling, SEO-optimized product content for online stores. "
|
||||
"Always respond with valid JSON containing the requested fields. "
|
||||
"HTML descriptions should use proper semantic HTML tags."
|
||||
)
|
||||
|
||||
user_msg = f"""Based on this product information:
|
||||
{context}
|
||||
|
||||
Generate the following content as a JSON object with these exact keys:
|
||||
|
||||
1. "title": {prompts.get('title', 'SEO-optimized product title in Title Case')}
|
||||
2. "short_description": {prompts.get('short_desc', 'Compelling 2-3 sentence HTML summary')}
|
||||
3. "long_description": {prompts.get('long_desc', 'Detailed HTML product description with headings and lists')}
|
||||
4. "meta_title": {prompts.get('meta_title', 'SEO meta title under 60 characters')}
|
||||
5. "meta_description": {prompts.get('meta_desc', 'SEO meta description under 160 characters')}
|
||||
6. "keywords": {prompts.get('keywords', 'Comma-separated SEO keywords')}
|
||||
|
||||
Respond ONLY with the JSON object, no markdown formatting."""
|
||||
|
||||
try:
|
||||
raw = self.generate(system, user_msg, max_tokens=3000)
|
||||
# Try to parse JSON from the response
|
||||
# Strip any markdown code fences
|
||||
cleaned = raw.strip()
|
||||
if cleaned.startswith('```'):
|
||||
cleaned = cleaned.split('\n', 1)[1]
|
||||
if cleaned.endswith('```'):
|
||||
cleaned = cleaned[:-3]
|
||||
cleaned = cleaned.strip()
|
||||
return json.loads(cleaned)
|
||||
except json.JSONDecodeError:
|
||||
_logger.warning("AI returned non-JSON response, returning raw text")
|
||||
return {
|
||||
'title': raw[:200] if raw else '',
|
||||
'short_description': '',
|
||||
'long_description': raw or '',
|
||||
'meta_title': '',
|
||||
'meta_description': '',
|
||||
'keywords': '',
|
||||
}
|
||||
|
||||
def generate_single_field(self, product_info, prompt, field_name):
|
||||
"""Generate a single field using the given prompt."""
|
||||
context = json.dumps(product_info, indent=2)
|
||||
system = (
|
||||
"You are an expert e-commerce copywriter and SEO specialist. "
|
||||
"Respond with ONLY the requested content, no explanations or formatting."
|
||||
)
|
||||
user_msg = f"Product info:\n{context}\n\nTask: {prompt}"
|
||||
|
||||
result = self.generate(system, user_msg, max_tokens=1500)
|
||||
return result.strip() if result else ''
|
||||
|
||||
def generate_image_metadata(self, product_name, product_category, prompt_alt, prompt_caption):
|
||||
"""Generate SEO metadata for a product image.
|
||||
|
||||
Returns:
|
||||
dict with: alt_text, caption, title, description
|
||||
"""
|
||||
system = (
|
||||
"You are an SEO specialist for e-commerce product images. "
|
||||
"Generate metadata that helps with image SEO and accessibility. "
|
||||
"Respond ONLY with a JSON object."
|
||||
)
|
||||
user_msg = f"""Product: {product_name}
|
||||
Category: {product_category}
|
||||
|
||||
Generate image metadata as JSON:
|
||||
- "alt_text": {prompt_alt} (under 125 characters)
|
||||
- "caption": {prompt_caption}
|
||||
- "title": SEO-optimized image title
|
||||
- "description": Descriptive image text for SEO
|
||||
|
||||
Respond ONLY with the JSON object."""
|
||||
|
||||
try:
|
||||
raw = self.generate(system, user_msg, max_tokens=500)
|
||||
cleaned = raw.strip()
|
||||
if cleaned.startswith('```'):
|
||||
cleaned = cleaned.split('\n', 1)[1]
|
||||
if cleaned.endswith('```'):
|
||||
cleaned = cleaned[:-3]
|
||||
cleaned = cleaned.strip()
|
||||
return json.loads(cleaned)
|
||||
except (json.JSONDecodeError, Exception):
|
||||
return {
|
||||
'alt_text': product_name,
|
||||
'caption': product_name,
|
||||
'title': product_name,
|
||||
'description': product_name,
|
||||
}
|
||||
@@ -1,104 +0,0 @@
|
||||
import base64
|
||||
import io
|
||||
import logging
|
||||
import struct
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ImageProcessor:
|
||||
"""Process product images: EXIF geo-tagging, metadata, optimization."""
|
||||
|
||||
@staticmethod
|
||||
def geo_tag_image(image_b64, company_name, address, phone, lat, lng):
|
||||
"""Write EXIF metadata with company info and GPS coordinates.
|
||||
|
||||
Args:
|
||||
image_b64: base64-encoded image data
|
||||
company_name: company name for Copyright/Artist tags
|
||||
address: company address for ImageDescription
|
||||
phone: company phone
|
||||
lat: GPS latitude (float)
|
||||
lng: GPS longitude (float)
|
||||
|
||||
Returns:
|
||||
base64-encoded image with EXIF data
|
||||
"""
|
||||
try:
|
||||
import piexif
|
||||
except ImportError:
|
||||
_logger.warning("piexif not installed — skipping geo-tagging. Run: pip install piexif")
|
||||
return image_b64
|
||||
|
||||
try:
|
||||
image_data = base64.b64decode(image_b64)
|
||||
|
||||
# Check if it's a JPEG (piexif only works with JPEG)
|
||||
if not image_data[:2] == b'\xff\xd8':
|
||||
_logger.info("Image is not JPEG — skipping EXIF geo-tagging")
|
||||
return image_b64
|
||||
|
||||
# Try to load existing EXIF
|
||||
try:
|
||||
exif_dict = piexif.load(image_data)
|
||||
except Exception:
|
||||
exif_dict = {"0th": {}, "Exif": {}, "GPS": {}, "1st": {}}
|
||||
|
||||
# Set 0th IFD tags
|
||||
exif_dict["0th"][piexif.ImageIFD.ImageDescription] = address.encode('utf-8') if address else b''
|
||||
exif_dict["0th"][piexif.ImageIFD.Copyright] = (
|
||||
f"Copyright {company_name}".encode('utf-8') if company_name else b''
|
||||
)
|
||||
exif_dict["0th"][piexif.ImageIFD.Artist] = company_name.encode('utf-8') if company_name else b''
|
||||
|
||||
# Set GPS tags if coordinates provided
|
||||
if lat and lng:
|
||||
lat_deg = ImageProcessor._decimal_to_dms(abs(lat))
|
||||
lng_deg = ImageProcessor._decimal_to_dms(abs(lng))
|
||||
|
||||
exif_dict["GPS"] = {
|
||||
piexif.GPSIFD.GPSLatitudeRef: b'N' if lat >= 0 else b'S',
|
||||
piexif.GPSIFD.GPSLatitude: lat_deg,
|
||||
piexif.GPSIFD.GPSLongitudeRef: b'E' if lng >= 0 else b'W',
|
||||
piexif.GPSIFD.GPSLongitude: lng_deg,
|
||||
}
|
||||
|
||||
exif_bytes = piexif.dump(exif_dict)
|
||||
output = io.BytesIO()
|
||||
piexif.insert(exif_bytes, image_data, output)
|
||||
return base64.b64encode(output.getvalue()).decode('utf-8')
|
||||
|
||||
except Exception as e:
|
||||
_logger.error("Failed to geo-tag image: %s", str(e))
|
||||
return image_b64
|
||||
|
||||
@staticmethod
|
||||
def _decimal_to_dms(decimal):
|
||||
"""Convert decimal degrees to EXIF DMS format (degrees, minutes, seconds as rationals)."""
|
||||
degrees = int(decimal)
|
||||
minutes_float = (decimal - degrees) * 60
|
||||
minutes = int(minutes_float)
|
||||
seconds = int((minutes_float - minutes) * 60 * 10000)
|
||||
return (
|
||||
(degrees, 1),
|
||||
(minutes, 1),
|
||||
(seconds, 10000),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def prepare_wc_image(image_b64, filename, alt_text='', caption='', title='', description=''):
|
||||
"""Prepare image data for WooCommerce upload.
|
||||
|
||||
Returns dict ready for WC product images array, using src as base64 data URL.
|
||||
Note: WC REST API v3 accepts image URLs in 'src'. For base64, we need to upload
|
||||
via WordPress media endpoint first, then reference by URL.
|
||||
"""
|
||||
return {
|
||||
'name': title or filename,
|
||||
'alt': alt_text or '',
|
||||
'caption': caption or '',
|
||||
'description': description or '',
|
||||
# The actual upload will be handled by the wizard
|
||||
'_base64': image_b64,
|
||||
'_filename': filename,
|
||||
}
|
||||
@@ -1,341 +0,0 @@
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
|
||||
import requests
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shared circuit breaker state — one per WC base URL so all Odoo workers
|
||||
# referencing the same WooCommerce store share the same breaker.
|
||||
# ---------------------------------------------------------------------------
|
||||
_circuit_breakers = {}
|
||||
_cb_lock = threading.Lock()
|
||||
|
||||
|
||||
class _CircuitBreaker:
|
||||
"""Per-host circuit breaker: CLOSED → OPEN after N failures, auto-resets
|
||||
after a cooldown period to HALF_OPEN (allows one probe request)."""
|
||||
|
||||
CLOSED = 'closed'
|
||||
OPEN = 'open'
|
||||
HALF_OPEN = 'half_open'
|
||||
|
||||
def __init__(self, failure_threshold=5, cooldown_seconds=60):
|
||||
self.failure_threshold = failure_threshold
|
||||
self.cooldown_seconds = cooldown_seconds
|
||||
self.state = self.CLOSED
|
||||
self.consecutive_failures = 0
|
||||
self.last_failure_time = 0
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def record_success(self):
|
||||
with self._lock:
|
||||
self.consecutive_failures = 0
|
||||
self.state = self.CLOSED
|
||||
|
||||
def record_failure(self):
|
||||
with self._lock:
|
||||
self.consecutive_failures += 1
|
||||
self.last_failure_time = time.monotonic()
|
||||
if self.consecutive_failures >= self.failure_threshold:
|
||||
self.state = self.OPEN
|
||||
_logger.warning(
|
||||
"Circuit breaker OPEN after %d consecutive failures — "
|
||||
"blocking requests for %ds",
|
||||
self.consecutive_failures, self.cooldown_seconds,
|
||||
)
|
||||
|
||||
def allow_request(self):
|
||||
with self._lock:
|
||||
if self.state == self.CLOSED:
|
||||
return True
|
||||
elapsed = time.monotonic() - self.last_failure_time
|
||||
if elapsed >= self.cooldown_seconds:
|
||||
self.state = self.HALF_OPEN
|
||||
_logger.info("Circuit breaker HALF_OPEN — allowing probe request")
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class _TokenBucket:
|
||||
"""Simple token-bucket rate limiter. Tokens refill at *rate* per second
|
||||
up to *capacity*. ``consume()`` blocks until a token is available."""
|
||||
|
||||
def __init__(self, rate, capacity):
|
||||
self.rate = rate
|
||||
self.capacity = capacity
|
||||
self.tokens = capacity
|
||||
self.last_refill = time.monotonic()
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def consume(self):
|
||||
while True:
|
||||
with self._lock:
|
||||
now = time.monotonic()
|
||||
elapsed = now - self.last_refill
|
||||
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
|
||||
self.last_refill = now
|
||||
if self.tokens >= 1:
|
||||
self.tokens -= 1
|
||||
return
|
||||
time.sleep(0.1)
|
||||
|
||||
|
||||
class WooApiClient:
|
||||
"""WooCommerce REST API v3 client wrapper with rate limiting and circuit
|
||||
breaker protection."""
|
||||
|
||||
# Default: 3 requests/sec, burst up to 5.
|
||||
# WooCommerce typically allows ~240 req/min (4/sec) so 3/sec is safe.
|
||||
DEFAULT_RATE = 3
|
||||
DEFAULT_BURST = 5
|
||||
|
||||
def __init__(self, url, consumer_key, consumer_secret,
|
||||
api_version='wc/v3', timeout=30,
|
||||
rate_limit=None, burst_limit=None):
|
||||
self.base_url = url.rstrip('/')
|
||||
self.api_version = api_version
|
||||
self.timeout = timeout
|
||||
|
||||
self.session = requests.Session()
|
||||
self.session.auth = (consumer_key, consumer_secret)
|
||||
self.session.headers.update({
|
||||
'Content-Type': 'application/json',
|
||||
'User-Agent': 'FusionWooCommerce/1.0',
|
||||
})
|
||||
|
||||
rate = rate_limit or self.DEFAULT_RATE
|
||||
burst = burst_limit or self.DEFAULT_BURST
|
||||
self._bucket = _TokenBucket(rate, burst)
|
||||
|
||||
with _cb_lock:
|
||||
if self.base_url not in _circuit_breakers:
|
||||
_circuit_breakers[self.base_url] = _CircuitBreaker(
|
||||
failure_threshold=5, cooldown_seconds=60,
|
||||
)
|
||||
self._breaker = _circuit_breakers[self.base_url]
|
||||
|
||||
def _url(self, endpoint):
|
||||
return f"{self.base_url}/wp-json/{self.api_version}/{endpoint}"
|
||||
|
||||
def _request(self, method, endpoint, data=None, params=None, retries=3):
|
||||
url = self._url(endpoint)
|
||||
|
||||
if not self._breaker.allow_request():
|
||||
raise ConnectionError(
|
||||
"WooCommerce API circuit breaker is OPEN for %s — "
|
||||
"too many consecutive failures. Retry later." % self.base_url
|
||||
)
|
||||
|
||||
last_exc = None
|
||||
for attempt in range(retries):
|
||||
self._bucket.consume()
|
||||
try:
|
||||
response = self.session.request(
|
||||
method, url,
|
||||
json=data, params=params,
|
||||
timeout=self.timeout,
|
||||
)
|
||||
|
||||
# --- Handle rate-limit response from WC / server ---
|
||||
if response.status_code == 429:
|
||||
retry_after = int(response.headers.get('Retry-After', 10))
|
||||
retry_after = min(retry_after, 120)
|
||||
_logger.warning(
|
||||
"WC API 429 on %s %s — backing off %ds (attempt %d/%d)",
|
||||
method, endpoint, retry_after, attempt + 1, retries,
|
||||
)
|
||||
time.sleep(retry_after)
|
||||
continue
|
||||
|
||||
# --- Non-retryable client errors (400-499 except 429) ---
|
||||
if 400 <= response.status_code < 500:
|
||||
_logger.error(
|
||||
"WC API %s %s returned %s (non-retryable): %s",
|
||||
method, endpoint, response.status_code,
|
||||
response.text[:500],
|
||||
)
|
||||
self._breaker.record_success()
|
||||
response.raise_for_status()
|
||||
|
||||
# --- Server errors (500+) are retryable ---
|
||||
if response.status_code >= 500:
|
||||
_logger.warning(
|
||||
"WC API %s %s returned %s (attempt %d/%d): %s",
|
||||
method, endpoint, response.status_code,
|
||||
attempt + 1, retries, response.text[:300],
|
||||
)
|
||||
last_exc = requests.HTTPError(response=response)
|
||||
wait = min(2 ** attempt * 2, 30)
|
||||
time.sleep(wait)
|
||||
continue
|
||||
|
||||
self._breaker.record_success()
|
||||
return response.json()
|
||||
|
||||
except requests.exceptions.ConnectionError as exc:
|
||||
last_exc = exc
|
||||
wait = min(2 ** attempt * 2, 30)
|
||||
_logger.warning(
|
||||
"WC API connection error %s %s (attempt %d/%d): %s — "
|
||||
"retrying in %ds",
|
||||
method, endpoint, attempt + 1, retries, exc, wait,
|
||||
)
|
||||
time.sleep(wait)
|
||||
|
||||
except requests.exceptions.Timeout as exc:
|
||||
last_exc = exc
|
||||
wait = min(2 ** attempt * 2, 30)
|
||||
_logger.warning(
|
||||
"WC API timeout %s %s (attempt %d/%d) — retrying in %ds",
|
||||
method, endpoint, attempt + 1, retries, wait,
|
||||
)
|
||||
time.sleep(wait)
|
||||
|
||||
except Exception as exc:
|
||||
last_exc = exc
|
||||
_logger.error(
|
||||
"WC API unexpected error %s %s: %s", method, endpoint, exc,
|
||||
)
|
||||
break
|
||||
|
||||
self._breaker.record_failure()
|
||||
raise last_exc
|
||||
|
||||
# --- Convenience methods ---
|
||||
|
||||
def get(self, endpoint, params=None):
|
||||
return self._request('GET', endpoint, params=params)
|
||||
|
||||
def post(self, endpoint, data):
|
||||
return self._request('POST', endpoint, data=data)
|
||||
|
||||
def put(self, endpoint, data):
|
||||
return self._request('PUT', endpoint, data=data)
|
||||
|
||||
def delete(self, endpoint):
|
||||
return self._request('DELETE', endpoint)
|
||||
|
||||
# --- Product endpoints ---
|
||||
|
||||
def get_products(self, page=1, per_page=100, **kwargs):
|
||||
params = {'page': page, 'per_page': per_page, **kwargs}
|
||||
return self.get('products', params=params)
|
||||
|
||||
def get_product(self, product_id):
|
||||
return self.get(f'products/{product_id}')
|
||||
|
||||
def get_product_variations(self, product_id, page=1, per_page=100):
|
||||
params = {'page': page, 'per_page': per_page}
|
||||
return self.get(f'products/{product_id}/variations', params=params)
|
||||
|
||||
def update_product(self, product_id, data):
|
||||
return self.put(f'products/{product_id}', data)
|
||||
|
||||
def create_product(self, data):
|
||||
return self.post('products', data)
|
||||
|
||||
# --- Attribute endpoints ---
|
||||
|
||||
def get_product_attributes(self):
|
||||
return self.get('products/attributes', params={'per_page': 100})
|
||||
|
||||
def create_product_attribute(self, data):
|
||||
return self.post('products/attributes', data)
|
||||
|
||||
def get_attribute_terms(self, attribute_id, page=1, per_page=100):
|
||||
return self.get(
|
||||
f'products/attributes/{attribute_id}/terms',
|
||||
params={'page': page, 'per_page': per_page},
|
||||
)
|
||||
|
||||
def create_attribute_term(self, attribute_id, data):
|
||||
return self.post(f'products/attributes/{attribute_id}/terms', data)
|
||||
|
||||
# --- Variation endpoints ---
|
||||
|
||||
def create_product_variation(self, product_id, data):
|
||||
return self.post(f'products/{product_id}/variations', data)
|
||||
|
||||
def update_product_variation(self, product_id, variation_id, data):
|
||||
return self.put(f'products/{product_id}/variations/{variation_id}', data)
|
||||
|
||||
def delete_product_variation(self, product_id, variation_id):
|
||||
return self.delete(f'products/{product_id}/variations/{variation_id}')
|
||||
|
||||
def batch_create_variations(self, product_id, variations_data):
|
||||
"""Create multiple variations at once using WC batch endpoint."""
|
||||
return self.post(
|
||||
f'products/{product_id}/variations/batch',
|
||||
{'create': variations_data},
|
||||
)
|
||||
|
||||
# --- Order endpoints ---
|
||||
|
||||
def get_orders(self, page=1, per_page=100, **kwargs):
|
||||
params = {'page': page, 'per_page': per_page, **kwargs}
|
||||
return self.get('orders', params=params)
|
||||
|
||||
def get_order(self, order_id):
|
||||
return self.get(f'orders/{order_id}')
|
||||
|
||||
def update_order(self, order_id, data):
|
||||
return self.put(f'orders/{order_id}', data)
|
||||
|
||||
# --- Customer endpoints ---
|
||||
|
||||
def get_customers(self, page=1, per_page=100, **kwargs):
|
||||
params = {'page': page, 'per_page': per_page, **kwargs}
|
||||
return self.get('customers', params=params)
|
||||
|
||||
def get_customer(self, customer_id):
|
||||
return self.get(f'customers/{customer_id}')
|
||||
|
||||
def create_customer(self, data):
|
||||
return self.post('customers', data)
|
||||
|
||||
def update_customer(self, customer_id, data):
|
||||
return self.put(f'customers/{customer_id}', data)
|
||||
|
||||
# --- Webhook endpoints ---
|
||||
|
||||
def create_webhook(self, data):
|
||||
return self.post('webhooks', data)
|
||||
|
||||
def get_webhooks(self):
|
||||
return self.get('webhooks', params={'per_page': 100})
|
||||
|
||||
def delete_webhook(self, webhook_id):
|
||||
return self.delete(f'webhooks/{webhook_id}')
|
||||
|
||||
# --- Tax endpoints ---
|
||||
|
||||
def get_tax_classes(self):
|
||||
return self.get('taxes/classes')
|
||||
|
||||
# --- Utility ---
|
||||
|
||||
def test_connection(self):
|
||||
try:
|
||||
result = self.get('system_status')
|
||||
wc_version = result.get('environment', {}).get('version', 'unknown')
|
||||
return True, wc_version
|
||||
except Exception as exc:
|
||||
return False, str(exc)
|
||||
|
||||
@staticmethod
|
||||
def verify_webhook_signature(payload, signature, secret):
|
||||
"""Verify a WooCommerce webhook HMAC-SHA256 signature."""
|
||||
if isinstance(payload, str):
|
||||
payload = payload.encode('utf-8')
|
||||
if isinstance(secret, str):
|
||||
secret = secret.encode('utf-8')
|
||||
computed = base64.b64encode(
|
||||
hmac.new(secret, payload, hashlib.sha256).digest()
|
||||
).decode('utf-8')
|
||||
return hmac.compare_digest(computed, signature)
|
||||
Reference in New Issue
Block a user