WooCommerceAPI与第三方同步

wen PHP项目 2026-07-22 186

本文目录导读：

核心同步方式
常见同步场景
数据映射配置
最佳实践建议

我来帮您梳理 WooCommerce API 与第三方系统同步的完整方案。

核心同步方式

REST API 直接同步

WooCommerce 原生提供 REST API：

// 基础配置
$woocommerce = new WooCommerce(
    'https://yourstore.com',
    'consumer_key',
    'consumer_secret',
    [
        'version' => 'wc/v3',
        'timeout' => 30,
    ]
);
// 获取订单
$orders = $woocommerce->get('orders', [
    'after' => '2024-01-01T00:00:00',
    'per_page' => 100
]);
// 创建产品
$product = $woocommerce->post('products', [
    'name' => '测试产品',
    'type' => 'simple',
    'regular_price' => '99.00',
    'description' => '产品描述'
]);

Webhook 实时推送

设置实时事件通知：

// 创建 Webhook
$webhook_data = [
    'name' => '订单同步',
    'topic' => 'order.created',
    'delivery_url' => 'https://your-system.com/webhook',
    'secret' => 'your_webhook_secret'
];
// 支持的 topics：
// order.created - 订单创建
// order.updated - 订单更新
// order.deleted - 订单删除
// product.created - 产品创建
// customer.created - 客户创建

中间件/消息队列方案

推荐使用 RabbitMQ 或 Redis：

# Python 示例：使用 Celery 处理同步
from celery import Celery
import requests
app = Celery('sync', broker='redis://localhost:6379/0')
@app.task
def sync_order_to_erp(order_id):
    # 获取WooCommerce订单
    order = get_woocommerce_order(order_id)
    # 转换数据格式
    erp_data = transform_order(order)
    # 发送到ERP系统
    response = requests.post(
        'https://erp-system.com/api/orders',
        json=erp_data
    )
    return response.status_code

常见同步场景

订单同步

# Python 示例：增量同步
def sync_orders(last_sync_time):
    import requests
    # WooCommerce API
    params = {
        'after': last_sync_time.isoformat(),
        'status': ['processing', 'completed'],
        'per_page': 100
    }
    response = requests.get(
        'https://yourstore.com/wp-json/wc/v3/orders',
        auth=('consumer_key', 'consumer_secret'),
        params=params
    )
    orders = response.json()
    for order in orders:
        # 转换订单格式
        synced_order = {
            'external_id': order['id'],
            'order_number': order['number'],
            'status': order['status'],
            'total': float(order['total']),
            'items': [{
                'product_id': item['product_id'],
                'quantity': item['quantity'],
                'price': float(item['price'])
            } for item in order['line_items']]
        }
        # 同步到第三方系统
        sync_to_third_party(synced_order)

库存同步

// PHP：库存双向同步
class StockSynchronizer {
    public function syncStock($product_id) {
        // 获取WooCommerce库存
        $wc_stock = $this->getWCStock($product_id);
        // 获取ERP库存
        $erp_stock = $this->getERPStock($product_id);
        // 比较并更新
        if ($wc_stock != $erp_stock) {
            // 根据优先级决定更新方向
            if ($this->wcPriority) {
                $this->updateERPStock($product_id, $wc_stock);
            } else {
                $this->updateWCStock($product_id, $erp_stock);
            }
        }
    }
    private function getWCStock($product_id) {
        // 通过API获取库存
    }
}

数据映射配置

{
  "field_mapping": {
    "customer": {
      "woocommerce": {
        "billing_first_name": "firstname",
        "billing_last_name": "lastname",
        "billing_email": "email",
        "billing_phone": "phone"
      },
      "erp_system": {
        "firstName": "firstname", 
        "lastName": "lastname",
        "emailAddress": "email",
        "phoneNumber": "phone"
      }
    },
    "product": {
      "woocommerce": {
        "sku": "sku",
        "name": "name",
        "regular_price": "price"
      },
      "third_party": {
        "itemCode": "sku",
        "description": "name", 
        "unitPrice": "price"
      }
    }
  }
}

最佳实践建议

错误处理与重试

class RetrySync:
    max_retries = 3
    retry_delay = 60  # seconds
    def sync_with_retry(self, data):
        for attempt in range(self.max_retries):
            try:
                return self.sync_data(data)
            except ConnectionError as e:
                if attempt < self.max_retries - 1:
                    time.sleep(self.retry_delay * (attempt + 1))
                    continue
                self.log_failure(data, str(e))

日志记录

// 同步日志记录
function log_sync_operation($operation, $status, $details) {
    global $wpdb;
    $wpdb->insert(
        'sync_logs',
        [
            'operation' => $operation,
            'status' => $status,
            'details' => json_encode($details),
            'created_at' => current_time('mysql')
        ]
    );
}