汤臣倍健营销云数据集成到金蝶云星辰V2：退货入库案例分享

在本次项目中，我们探讨的是如何将汤臣倍健营销云系统中的销售退货订单数据，快速、安全地集成至金蝶云星辰V2系统，实现全自动化的退货入库管理。该方案被命名为“退货入库(杭州汤臣倍健)”，旨在通过使用轻易云数据集成平台对接双方系统，达到无缝连接和高效同步的数据处理效果。

确保不漏单的关键措施

为了确保从汤臣倍健营销云获取的数据不会出现遗漏情况，我们利用了其提供的/erp/api/order/query/saleReturnOrder接口，通过定时任务可靠抓取接口数据。同时，通过分页机制解决限流问题，即每次提取一小部分数据直到所有记录都被获取。这种方式不仅避免了API调用超出限制，还保证了大规模数据能持续稳定地传输至目标系统。

高效写入与批量处理

大量返回的销售退货订单需要及时、准确地写入到金蝶云星辰V2，这里我们用到了/jdy/v2/scm/sal_in_bound接口实现批量写入。通过合理设计的数据映射与流程控制，将不同来源格式的数据进行标准转换，并批量推送到金蝶仓储模块，使得整个流程顺畅且高效。

数据对接注意事项及异常处理

在对接过程中，需要特别关注两点：首先是两个系统之间字段和格式差异，如时间格式、金额显示等细节；其次是在实施过程中，有可能遇到网络抖动或服务中断，为此制定了一套完善的异常重试机制，当发现调用失败或超时报错时，会自动重试并记录日志，以便后续排查治理，保证业务不中断。

总之，通过全面考虑以上多个方面的问题，本次案例展示的不仅是技术上的具体实现，更是一种稳妥、高效的数据处理思路。在详细实现方案部分将进一步展开各个步骤的方法和代码示例，以供参考借鉴。

调用源系统汤臣倍健营销云接口获取并加工数据

在数据集成的生命周期中，调用源系统接口是至关重要的一步。本文将深入探讨如何通过轻易云数据集成平台调用汤臣倍健营销云的/erp/api/order/query/saleReturnOrder接口，获取并加工退货入库数据。

接口调用配置

首先，我们需要配置API接口的元数据，以确保能够正确地请求和处理数据。以下是该接口的元数据配置：

{
  "api": "/erp/api/order/query/saleReturnOrder",
  "effect": "QUERY",
  "method": "POST",
  "number": "number",
  "id": "id",
  "idCheck": true,
  "request": [
    {"field":"tenantId","label":"经销商id","type":"string","describe":"经销商id(必填,营销云id)如：34cc4109705e4c058b7b3b0352e57d31","value":"xxx"},
    {"field":"yxyNumber","label":"营销云销售订单号","type":"string","describe":"如：YD1215710122031701，传此参数时，其他时间状态等条件无效"},
    {"field":"number","label":"系统订单号","type":"string","describe":"如：XOUT0000000293，传此参数时,其他时间状态等条件无效"},
    {"field":"status","label":"订单状态","type":"string","describe":"0:未审核 1：已审核（已出库）","value":"1"},
    {"field":"beginTime","label":"开始时间","type":"string","describe":"timeType为空或者0，基于创建时间查询，timeType为1，基于更新时间查询，格式：0000-00-00或0000-00-00 00:00:00，如果不传单号此字段必填","value":"{{LAST_SYNC_TIME|datetime}}"},
    {"field":"endTime","label":"结束时间","type":"string","describe":"timeType为空或者0，基于创建时间查询，timeType为1，基于更新时间查询，格式：0000-00-00或0000-00-00 00:00:00，如果不传单号此字段必填","value":"{{CURRENT_TIME|datetime}}"},
    {"field":"pageNo","label":"页码","type":"string","describe":"默认1","value":"1"},
    {"field":"pageSize","label":"每页条数","type":"string","describe":"默认30","value":"30"},
    {"field":"timeType","label":"时间段标志","type":"string","describe":"查询时间段标识，0:创建时间（默认），1：最后更新时间","value":""}
  ],
  "autoFillResponse": true
}

请求参数详解

在请求参数中，有几个关键字段需要特别注意：

• tenantId: 经销商ID，这是一个必填项，用于标识具体的经销商。
• yxyNumber 和 number: 分别代表营销云销售订单号和系统订单号，这两个字段用于精确查询特定订单。
• status: 表示订单状态，我们通常设置为1表示已审核（已出库）。
• beginTime 和 endTime: 用于定义查询的时间范围，这两个字段在没有传递订单号时是必填项。
• pageNo 和 pageSize: 分别表示分页查询的页码和每页条数。

数据请求与清洗

在发送请求后，我们会接收到原始的JSON格式数据。为了便于后续处理，我们需要对这些数据进行清洗和转换。以下是一个简单的数据清洗示例：

import requests
import json
from datetime import datetime

# 定义请求URL和头信息
url = 'https://api.example.com/erp/api/order/query/saleReturnOrder'
headers = {'Content-Type': 'application/json'}

# 定义请求体
payload = {
    'tenantId': '34cc4109705e4c058b7b3b0352e57d31',
    'status': '1',
    'beginTime': (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d %H:%M:%S'),
    'endTime': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
    'pageNo': '1',
    'pageSize': '30',
}

# 发起POST请求
response = requests.post(url, headers=headers, data=json.dumps(payload))

# 检查响应状态码
if response.status_code == 200:
    data = response.json()

    # 数据清洗示例
    cleaned_data = []
    for order in data['orders']:
        cleaned_order = {
            'order_id': order['id'],
            'order_number': order['number'],
            'order_status': order['status'],
            # 添加更多需要清洗和转换的字段
        }
        cleaned_data.append(cleaned_order)

    # 输出清洗后的数据
    print(cleaned_data)
else:
    print(f"Error: {response.status_code}")

数据转换与写入

经过清洗的数据需要转换成目标系统所需的格式，并写入到相应的数据存储中。这一步通常涉及到对字段名、数据类型等进行调整，以确保兼容性。

例如，将清洗后的数据写入到数据库中：

import sqlite3

# 创建数据库连接
conn = sqlite3.connect('example.db')
cursor = conn.cursor()

# 创建表结构（如果不存在）
cursor.execute('''
CREATE TABLE IF NOT EXISTS sale_return_orders (
    order_id TEXT PRIMARY KEY,
    order_number TEXT,
    order_status TEXT
)
''')

# 插入清洗后的数据
for order in cleaned_data:
    cursor.execute('''
        INSERT OR REPLACE INTO sale_return_orders (order_id, order_number, order_status)
        VALUES (?, ?, ?)
        ''', (order['order_id'], order['order_number'], order['order_status']))

# 提交事务并关闭连接
conn.commit()
conn.close()

通过以上步骤，我们实现了从汤臣倍健营销云接口获取退货入库数据，并对其进行清洗、转换和写入数据库的全过程。这不仅提高了数据处理效率，也确保了业务流程的顺畅运行。

退货入库数据集成到金蝶云星辰V2API接口的ETL转换

在轻易云数据集成平台上，将源平台的数据进行ETL转换并写入目标平台金蝶云星辰V2API接口，是一个复杂且精细的过程。本文将详细探讨如何利用元数据配置，完成从源数据到目标API格式的转换和写入。

数据请求与清洗

在生命周期的第一阶段，我们已经完成了数据的请求与清洗。现在，我们进入第二阶段，即将清洗后的数据进行ETL转换，并通过API接口写入金蝶云星辰V2系统。

元数据配置解析

根据提供的元数据配置，我们需要将源平台的数据转换为金蝶云星辰V2 API所需的格式。以下是关键字段及其对应关系：

• bill_source: 固定值 "ISV"
• bill_date: 出库日期，格式为 YYYY-MM-DD，通过模板变量 {{auditTime|date}} 获取。
• customer_id: 客户ID，通过 _findCollection 方法从特定集合中查找。
• remark: 单据备注，由多个字段拼接而成。
• contact_address, contact_linkman, contact_phone: 联系信息相关字段，直接映射。
• dept_number: 部门编码，固定值 "BM00001"。
• material_entity: 商品分录数组，包含多个子字段。

数据转换逻辑

1. 固定值和模板变量

   • bill_source 和 dept_number 是固定值，可以直接赋值。
   • bill_date 使用模板变量 {{auditTime|date}} 进行日期格式化。

2. 查找和映射

   • customer_id 通过 _findCollection find id from b41660e7-fa00-318f-bbee-1395e229ee6b where number={extCusCode} 查找客户ID。
   • 商品分录中的 material_id 和 unit_id 也需要通过 _findCollection 方法查找。

3. 数组处理

   • material_entity 是一个数组，每个元素包含多个子字段，如 material_id, stock_id, qty, 等。每个子字段可能需要进一步处理，如查找或格式化。

示例代码

以下是一个简化的示例代码，用于演示如何将源数据转换为目标API所需的格式：

{
  "api": "/jdy/v2/scm/sal_in_bound",
  "method": "POST",
  "data": {
    "bill_source": "ISV",
    "bill_date": "{{auditTime|date}}",
    "customer_id": "_findCollection find id from b41660e7-fa00-318f-bbee-1395e229ee6b where number={extCusCode}",
    "remark": "{remark}-来自营销云-{number}",
    "contact_address": "{shippingAddress}",
    "contact_linkman": "{contacts}",
    "contact_phone": "{phone}",
    "dept_number": "BM00001",
    "material_entity": [
      {
        "material_id": "_findCollection find id from 395d1591-aee0-3f10-ad09-d5593167af48 where number={extMaterialNo}",
        "stock_id": "1320317872534215680",
        "qty": "{{itemList.opernumber}}",
        "tax_price": "{{itemList.taxunitprice}}",
        "is_free": false,
        "unit_id": "_findCollection find base_unit_id from 395d1591-aee0-3f10-ad09-d5593167af48 where number={extMaterialNo}",
        "batch_no": "{{itemList._Flot}}",
        "comment": "{{itemList.remark}}",
        "kf_date": "{{itemList._Fmfg}}",
        "valid_date": "{{itemList._Fexp}}",
        "kf_period": "730",
        "kf_type": "1"
      }
      // ... more items
    ]
  }
}

数据写入

一旦数据按照上述逻辑转换完毕，即可通过HTTP POST请求将其写入金蝶云星辰V2 API接口。确保请求头中包含必要的认证信息，并正确处理响应结果，以便监控和调试。

import requests
import json

url = 'https://api.kingdee.com/jdy/v2/scm/sal_in_bound'
headers = {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer YOUR_ACCESS_TOKEN'
}

data = {
    # 上述JSON结构
}

response = requests.post(url, headers=headers, data=json.dumps(data))
if response.status_code == 200:
    print('Data successfully written to Kingdee Cloud.')
else:
    print(f'Failed to write data: {response.text}')

通过上述步骤，我们可以实现从源平台到金蝶云星辰V2 API接口的数据无缝对接。这不仅提高了数据处理效率，还确保了数据的一致性和准确性。