Odoo-Modules/fusion-woo-odoo/fusion_woocommerce/lib/woo_api_client.py

import base64
import hashlib
import hmac
import logging
import time

import requests

_logger = logging.getLogger(__name__)


class WooApiClient:
    """WooCommerce REST API v3 client wrapper."""

    def __init__(self, url, consumer_key, consumer_secret, api_version='wc/v3', timeout=30):
        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',
        })

    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)
        last_exc = None
        for attempt in range(retries):
            try:
                response = self.session.request(
                    method,
                    url,
                    json=data,
                    params=params,
                    timeout=self.timeout,
                )
                response.raise_for_status()
                return response.json()
            except Exception as exc:
                last_exc = exc
                wait = 2 ** attempt
                _logger.warning(
                    "WooCommerce API %s %s failed (attempt %d/%d): %s — retrying in %ds",
                    method, endpoint, attempt + 1, retries, exc, wait,
                )
                if attempt < retries - 1:
                    time.sleep(wait)
        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.

        Args:
            payload (bytes): Raw request body bytes.
            signature (str): Value of the X-WC-Webhook-Signature header.
            secret (str): The webhook secret configured in WooCommerce.

        Returns:
            bool: True if the signature matches, False otherwise.
        """
        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)