erp-integration

ERP 集成

概述

为电商平台与 ERP 系统（SAP、NetSuite、Odoo、Microsoft Dynamics）构建集成，实现订单、库存、客户和产品的双向同步。本技能涵盖集成架构模式（事件驱动、轮询、中间件）、数据映射、冲突解决、采用重试策略的错误处理，以及可防止重复记录的幂等同步。

何时使用此技能

• 将在线商店连接到 ERP，以实现自动化订单履约时
• 将 ERP/WMS 中的实时库存水平同步到电商目录时
• 在在线商店与 ERP 之间构建客户主数据同步时
• 实现从 ERP 到在线商店的产品和定价数据源时
• 设计用于处理多个集成点的中间件层时

核心说明

步骤 1：确定平台和集成方式

平台	集成选项	推荐方式
Shopify	Shopify Admin API + 用于订单数据的 Webhook	使用 Zapier（无代码，$20/月）或 Celigo（iPaaS）连接标准 ERP；对于 NetSuite，使用官方的 NetSuite Connector for Shopify 应用；有定制需求时使用 Shopify Webhook
WooCommerce	WooCommerce REST API + WordPress 钩子	简单流程可使用 Zapier；安装 Zynk WooCommerce 连接器（售价 £500 起）以连接 SAP/NetSuite；或者使用 WooCommerce REST API 构建自定义集成
BigCommerce	BigCommerce API + Webhook	使用 Celigo 或 Boomi 连接企业级 ERP；BigCommerce 的应用市场中提供适用于 NetSuite、SAP 和 Microsoft Dynamics 的预构建连接器
自定义 / 无头架构	完整 API 访问权限——构建中间件服务	实现事件驱动的订单同步（商店 Webhook → 队列 → ERP 适配器）、基于轮询的库存同步（定时任务 → ERP API → 更新目录），并为失败的同步设置死信队列

步骤 2：针对平台的 ERP 集成

---

Shopify

使用预构建连接器连接标准 ERP：

1. 对于 NetSuite：从 Shopify App Store 安装官方 NetSuite Connector for Shopify（$150-$300/月）。它可以双向同步订单、库存和客户，无需编写自定义代码
2. 对于 SAP：使用 Celigo 的 SAP + Shopify 集成模板，或联系你的 SAP 合作伙伴获取其 Shopify 连接器
3. 对于 Odoo：在 Odoo 中安装 Odoo Shopify Connector 模块（免费，社区版）——在 Odoo 设置中配置 Shopify API 凭据

使用 Shopify Webhook 连接自定义 ERP：

1. 在 Shopify 管理后台中，前往 设置 → 通知 → Webhook
2. 点击 创建 Webhook，并为 orders/create、orders/paid 和 inventory_levels/update 添加端点
3. 你的 ERP 中间件端点会以 JSON 格式接收订单数据，并将其转换为 ERP 所需的格式
4. 如需将库存同步回 Shopify，请在每次 ERP 轮询后使用 Inventory API 更新库存水平

---

WooCommerce

使用 Zapier 实现简单、低流量的 ERP 同步：

1. 在 zapier.com 中连接 WooCommerce 和你的 ERP（NetSuite、Odoo、Sage）
2. 创建一个 Zap：触发器 = WooCommerce 中的新订单，操作 = 在 NetSuite 中创建销售订单
3. 在 Zapier 的字段映射器中，将 WooCommerce 订单字段映射到 ERP 所需字段
4. 免费套餐中的 Zapier 每 15 分钟轮询一次 WooCommerce；升级到 Starter（$19.99/月）即可加快轮询频率

对于更高流量或自定义 ERP 集成：

1. 安装 WP Webhooks（免费，wordpress.org），将 WooCommerce 事件发送到你的中间件
2. 使用 WooCommerce REST API（/wp-json/wc/v3/orders）和 OAuth 1.0a，让中间件拉取订单
3. 如需将 ERP 库存同步到 WooCommerce，请使用 WooCommerce Products API（PUT /wp-json/wc/v3/products/{id}）更新 stock_quantity

---

自定义 / 无头

集成架构 — 选择适合你的 ERP 的模式：

Event-Driven (recommended for real-time order sync):
  Storefront → Webhook/Event → Message Queue (SQS/BullMQ) → ERP Adapter

Polling (for ERPs without webhooks, like legacy SAP installations):
  Scheduler → Poll ERP API → Transform → Update Storefront

Middleware Platform (for complex multi-system environments):
  Storefront ↔ Celigo / MuleSoft / Workato ↔ ERP

幂等订单同步服务：

// lib/erp/order-sync.ts
export async function syncOrder(orderId: string): Promise<void> {
  const order = await db.orders.getWithItems(orderId);

  // Check if already synced
  const existingSync = await db.syncLog.findByOrderId(orderId);
  if (existingSync?.status === 'synced') return;

  // Check ERP for existing record (guards against retries after partial failure)
  const existing = await erpAdapter.findOrderByExternalReference(order.orderNumber);
  if (existing) {
    await db.syncLog.upsert({ orderId, externalId: existing.erpOrderId, status: 'synced' });
    return;
  }

  try {
    const erpOrder = mapOrderToERP(order);
    const { erpOrderId } = await erpAdapter.createSalesOrder(erpOrder);
    await db.syncLog.upsert({ orderId, externalId: erpOrderId, status: 'synced', syncedAt: new Date() });
    await db.orders.updateMetadata(orderId, { erpOrderId });
  } catch (error) {
    await db.syncLog.upsert({ orderId, status: 'failed', lastError: error.message });
    throw error; // Let the retry mechanism handle it
  }
}

function mapOrderToERP(order: Order): ERPSalesOrder {
  return {
    externalReference: order.orderNumber,
    orderDate: order.createdAt.toISOString().split('T')[0],
    customer: {
      externalId: order.customer?.erpCustomerId || null,
      email: order.email,
      name: `${order.shippingAddress.firstName} ${order.shippingAddress.lastName}`,
    },
    shippingAddress: {
      line1: order.shippingAddress.street1,
      city: order.shippingAddress.city,
      state: order.shippingAddress.state,
      postalCode: order.shippingAddress.postalCode,
      country: order.shippingAddress.country,
    },
    lineItems: order.lineItems.map(item => ({
      sku: item.sku,
      quantity: item.quantity,
      unitPrice: item.unitPrice / 100,   // Convert cents to dollars for ERP
      taxAmount: item.taxAmount / 100,
    })),
    orderTotal: order.totalPrice / 100,
    currency: order.currency,
  };
}

使用 BullMQ 队列实现可靠的订单同步，并通过指数退避进行重试：

import { Queue, Worker, QueueEvents } from 'bullmq';

const orderSyncQueue = new Queue('order-sync', {
  connection: { host: process.env.REDIS_HOST, port: 6379 },
  defaultJobOptions: {
    attempts: 5,
    backoff: { type: 'exponential', delay: 5000 }, // 5s, 10s, 20s, 40s, 80s
    removeOnComplete: { count: 1000 },
    removeOnFail: { count: 5000 },
  },
});

// Producer: enqueue when order is placed
export async function onOrderPlaced(orderId: string) {
  await orderSyncQueue.add('sync-order', { orderId }, {
    jobId: `order-sync-${orderId}`, // Prevents duplicate queue entries
  });
}

// Consumer: process sync
new Worker('order-sync', async (job) => {
  await syncOrder(job.data.orderId);
}, { connection: { host: process.env.REDIS_HOST, port: 6379 }, concurrency: 5 });

库存同步（基于轮询，从 ERP 同步到店面）：

// lib/erp/inventory-sync.ts — run via scheduled job every 5 minutes
export async function syncInventoryLevels(): Promise<void> {
  const lastSyncAt = await redis.get('erp:inventory:last_sync');
  let page = 1, hasMore = true;

  while (hasMore) {
    const { items, hasMore: more } = await erpAdapter.getInventoryLevels({
      page, pageSize: 500,
      modifiedSince: lastSyncAt ? new Date(lastSyncAt) : undefined,
    });

    for (const item of items) {
      // Available = On Hand - Reserved - Safety Stock
      const available = Math.max(0, item.onHandQuantity - item.reservedQuantity - (item.safetyStock || 0));
      const current = await db.inventory.getQuantityBySku(item.sku);
      if (current === available) continue; // Skip unchanged

      await db.inventory.updateBySku(item.sku, { quantity: available, lastSyncedAt: new Date() });
      // Update Redis cache for real-time product page availability
      const productId = await db.inventory.getProductIdBySku(item.sku);
      if (productId) await redis.setex(`inventory:${productId}`, 3600, String(available));
    }

    hasMore = more;
    page++;
  }

  await redis.set('erp:inventory:last_sync', new Date().toISOString());
}

最佳实践

• 确保每项同步操作都具备幂等性 — 使用外部引用（订单号、SKU）检查 ERP 中是否已有记录，再创建记录；这样可以防止重试导致重复
• 使用消息队列同步订单 — 切勿在结账时同步调用 ERP；应将任务加入队列并异步处理，同时进行重试
• 实现死信队列 — 所有重试均告失败后，将失败任务移至 DLQ 以供人工检查；切勿静默丢弃消息
• 使用增量同步，而非全量同步 — 查询 ERP 中自上次同步时间戳以来有修改的记录；全量同步在记录超过几千条后无法扩展
• 在本地记录中保存 ERP 记录 ID — 订单同步后，将 ERP 订单 ID 保存到本地记录中，以便交叉引用并在日后查询状态
• 正确计算可用库存 — available = onHand - reserved - safetyStock，并将结果限制为不小于零；切勿直接使用 ERP 中的现有库存数量

常见问题

问题	解决方案
重试逻辑导致 ERP 中出现重复订单	使用电商订单号作为外部引用，并在创建前检查其是否存在；大多数 ERP 都支持通过外部 ID 检查重复项
同步后库存数量变为负数	将可用数量限制为不小于零；Math.max(0, onHand - reserved - safetyStock)
ERP 的速率限制导致同步失败	为每项操作实现速率限制器（令牌桶），大批量同步时使用 ERP 的批量 / Feed API，而不是逐项调用
价格同步覆盖促销价格	将基础价格（来自 ERP）与促销价格（由电商平台管理）分开；切勿让 ERP 同步覆盖正在生效的促销价格
大批量初始数据加载超时	将初始同步拆分为多个批次并设置检查点，以便失败后恢复；使用并发限制并行处理

相关技能

• @webhook-architecture
• @marketplace-connectors
• @monitoring-alerting-commerce
• @product-information-management