### 汤臣倍健营销云数据集成到金蝶云星辰V2：采购入库同步案例分享
在本篇技术案例中，我们将详细探讨如何通过轻易云数据集成平台，实现汤臣倍健营销云与金蝶云星辰V2之间的无缝对接，完成“采购入库同步（Life-Space）杭州亿倍盛”任务。该方案旨在确保从汤臣倍健营销云获取的采购入库单据能够可靠、高效地写入到金蝶云星辰V2系统中，从而提升企业整体的数据处理效率和业务响应速度。
为了实现这一目标，我们主要采用了以下关键技术措施：
1. **高吞吐量的数据写入能力**：
借助轻易云数据集成平台的强大处理能力，大量来自汤臣倍健营销云的采购入库数据能够以高吞吐量迅速被写入到金蝶云星辰V2。这一特性极大减少了数据传输时延，使得业务处理更加及时。
2. **API接口调用及分页限流处理**：
在整个集成过程中，通过调用汤臣倍健营销云提供的`/erp/api/order/query/purInWarehsOrder` API接口实时抓取数据，并结合分页与限流机制，确保每次请求都能有效获取所需信息，同时避免因大量请求导致服务器负载过重的问题。
3. **定制化的数据转换逻辑**：
为适应两套系统间不同的数据结构需求，在将获取到的数据推送至金蝶云星辰V2时，使用定制化的数据映射和转换逻辑，将其格式规范到符合目的端要求。一方面，这种灵活设置使得各种复杂情况下的数据均能准确转移；另一方面，这也使我们能够通过向`/jdy/v2/scm/pur_inbound` API接口成功进行批量提交，让所有相关联的信息得到完整维护与更新。
4. **实时监控和异常检测**:
通过集中监控系统，不仅可以实时追踪每个数据集成任务的状态，还支持异常检测，如果出现错误（例如网络波动或服务器故障），自动触发告警并执行重试机制，以保证后续操作不中断。此外，该平台记录所有日志便于日后审计和分析问题来源，从根本上提高整体可靠性。
### 调用源系统汤臣倍健营销云接口/erp/api/order/query/purInWarehsOrder获取并加工数据
在数据集成生命周期的第一步，我们需要从源系统获取数据，并对其进行初步加工。本文将详细探讨如何通过调用汤臣倍健营销云接口`/erp/api/order/query/purInWarehsOrder`来实现这一目标。
#### 接口调用配置
首先，了解接口的基本配置是至关重要的。该接口采用POST方法进行数据请求，主要用于查询采购入库订单。以下是元数据配置中的关键字段及其描述：
- **tenantId**: 经销商id（必填），如：08e2fd9d074f4ef7b5e6416b74d006d6
- **yxyNumber**: 营销云销售订单号，如：lzwhy0121012002
- **number**: 系统订单号，如：CGDD00000003744
- **status**: 订单状态，0表示未审核，1表示已审核（已出库）
- **beginTime**: 开始时间，格式为0000-00-00或0000-00-0000:00:00
- **endTime**: 结束时间，格式同上
- **pageNo**: 页码，默认值为1
- **pageSize**: 每页条数，默认值为30
- **timeType**: 时间段标志，0表示创建时间（默认），1表示最后更新时间
#### 请求参数构建
根据元数据配置，我们需要构建一个POST请求的JSON体。以下是一个示例请求体：
```json
{
"tenantId": "08e2fd9d074f4ef7b5e6416b74d006d6",
"yxyNumber": "",
"number": "",
"status": "1",
"beginTime": "{{LAST_SYNC_TIME|datetime}}",
"endTime": "{{CURRENT_TIME|datetime}}",
"pageNo": "1",
"pageSize": "30",
"timeType": "1"
}
```
在这个请求体中，我们使用了动态变量`{{LAST_SYNC_TIME|datetime}}`和`{{CURRENT_TIME|datetime}}`来分别表示上次同步时间和当前时间。这些变量将在实际请求时被替换为具体的日期时间值。
#### 数据清洗与条件过滤
获取到原始数据后，需要对其进行清洗和过滤。根据元数据配置中的条件，我们需要排除掉物料编号包含"F.A"的记录：
```json
"condition":[[{"field":"itemList.materialNumber","logic":"notlike","value":"F.A"}]]
```
这意味着我们需要遍历返回的数据列表，并移除所有物料编号包含"F.A"的项。以下是一个简单的数据清洗示例：
```python
def clean_data(data):
cleaned_data = []
for item in data:
if not any("F.A" in material['materialNumber'] for material in item['itemList']):
cleaned_data.append(item)
return cleaned_data
```
#### 自动填充响应
元数据配置中还提到自动填充响应`autoFillResponse:true`，这意味着在处理完数据后，我们可以直接将其写入目标系统，而无需手动映射字段。这大大简化了数据集成过程，提高了效率。
#### 实践案例
假设我们已经成功调用了API并获得了如下原始响应数据：
```json
[
{
"id": "12345",
"number": "CGDD00000003744",
"status": "1",
"itemList": [
{"materialNumber": "A123"},
{"materialNumber": "B456"}
]
},
{
"id": "67890",
"number": "CGDD00000003745",
"status": "1",
"itemList": [
{"materialNumber": "F.A789"},
{"materialNumber": "C789"}
]
}
]
```
经过清洗后的数据应如下所示：
```json
[
{
"id": "12345",
"number": "CGDD00000003744",
"status": "1",
"itemList": [
{"materialNumber": "A123"},
{"materialNumber": "B456"}
]
}
]
```
通过上述步骤，我们成功地从汤臣倍健营销云接口获取并加工了采购入库订单的数据，为后续的数据转换与写入奠定了基础。
### 使用轻易云数据集成平台将采购入库数据转换并写入金蝶云星辰V2API接口
在数据集成的生命周期中，ETL（Extract, Transform, Load）转换是关键步骤之一。本文将重点探讨如何通过轻易云数据集成平台，将采购入库的数据转换为金蝶云星辰V2API接口所能接收的格式，并最终写入目标平台。
#### API接口配置及元数据解析
在进行ETL转换之前，首先需要了解目标API接口的具体配置和要求。以下是金蝶云星辰V2的采购入库接口配置：
```json
{
"api": "/jdy/v2/scm/pur_inbound",
"effect": "EXECUTE",
"method": "POST",
"number": "1",
"id": "1",
"name": "1",
"idCheck": true,
"request": [
{"field":"bill_date","label":"单据日期","type":"string","describe":"单据日期","value":"{{auditTime|date}}"},
{"field":"bill_no","label":"单据编码","type":"string","describe":"单据编码"},
{"field":"transtype","label":"业务类型","type":"string","value":"2"},
{"field":"supplier_id","label":"供应商编码","type":"string","describe":"供应商编码","value":"1320329187667502080"},
{"field":"remark","label":"备注","type":"string","describe":"商品编码","value":"{remark}-来自营销云-{yxyNumber}"},
{
"field": "material_entity",
"label": "商品分录",
"type": "array",
"describe": "含税单价",
"value": "itemList",
"children": [
{"field":"material_number","label":"商品编码","type":"string","describe":"商品编码","value":"{extMaterialNo}"},
{"field":"stock_id","label":"仓库编码","type":"string","describe":"仓库编码","value":"1320321537273803776"},
{"field":"is_free","label":"是否赠品","type":"int","describe":"是否赠品","value":"false"},
{"field":"qty","label":"数量","type":"string","describe":"数量","value":"{{itemList.opernumber}}"},
{"field": "unit_id",
"label": "单位",
"type": "string",
"describe": "单位",
"value": "_findCollection find base_unit_id from 395d1591-aee0-3f10-ad09-d5593167af48 where number={extMaterialNo}"
},
{"field": "batch_no",
"label": "批次号",
"type": "string",
"value": "{{itemList._Flot}}"
},
{"field":
kf_date,
label: 生产日期,
type: string,
value: {{itemList._Fmfg}}
},
{"field":
kf_period,
label: 保质期天数,
type: string,
value: 730
},
{"field":
valid_date,
label: 有效日期,
type: string,
value: {{itemList._Fexp}}
},
{"field":
kf_type,
label: 保质期类型,
type: string,
describe: 1/天，2/月，3/年（保质期商品必填）,
value: 1
},
{"field":
tax_price,
label: 含税单价,
type: string,
describe: 含税单价,
value: _function {{itemList.taxlastmoney}}/{{itemList.opernumber}}
},
{"field":
cess,
label: 税率,
type: string,
describe: 税率
},
{
field:
all_amount,
label:
价税合计,
type:
string,
value:
{{itemList.taxlastmoney}}
}
]
}
]
}
```
#### 数据转换过程
在ETL过程中，我们需要对源平台的数据进行清洗和转换，以符合目标API接口的要求。以下是具体步骤：
1. **提取数据**：从源平台提取采购入库相关的数据，包括单据日期、单据编码、供应商信息、商品信息等。
2. **数据清洗**：对提取的数据进行必要的清洗和校验。例如，将日期格式化为目标平台所需的格式，将供应商编码映射为目标平台的供应商ID等。
3. **数据转换**：
- **单据日期**：使用模板变量`{{auditTime|date}}`将审核时间格式化为字符串。
- **单据编码**：直接映射源平台的单据编码。
- **业务类型**：固定值“2”表示采购入库。
- **供应商编码**：固定值“1320329187667502080”。
- **备注**：使用模板变量`{remark}-来自营销云-{yxyNumber}`生成备注信息。
- **商品分录**：
- 商品编码、仓库编码、是否赠品、数量、单位、批次号、生产日期、保质期天数、有效日期、保质期类型、含税单价、税率和价税合计等字段分别根据元数据配置进行映射和计算。
4. **数据加载**：将转换后的数据通过POST请求写入金蝶云星辰V2API接口。
#### 示例代码
以下是一个简化的示例代码，展示了如何使用轻易云数据集成平台进行上述操作：
```python
import requests
import json
# 假设已经从源平台提取了原始数据
source_data = {
# 示例原始数据
}
# 转换后的目标数据结构
target_data = {
'bill_date': source_data['auditTime'], # 格式化后的日期
'bill_no': source_data['billNo'],
'transtype': '2',
'supplier_id': '1320329187667502080',
'remark': f"{source_data['remark']}-来自营销云-{source_data['yxyNumber']}",
'material_entity': []
}
# 转换商品分录信息
for item in source_data['itemList']:
material_entry = {
'material_number': item['extMaterialNo'],
'stock_id': '1320321537273803776',
'is_free': False,
'qty': item['opernumber'],
'unit_id': find_unit_id(item['extMaterialNo']),
'batch_no': item['_Flot'],
'kf_date': item['_Fmfg'],
'kf_period': '730',
'valid_date': item['_Fexp'],
'kf_type': '1',
'tax_price': calculate_tax_price(item),
'cess': item['taxRate'],
'all_amount': item['taxlastmoney']
}
target_data['material_entity'].append(material_entry)
# 将目标数据发送到金蝶云星辰V2API接口
response = requests.post(
url="https://api.kingdee.com/jdy/v2/scm/pur_inbound",
headers={"Content-Type": "application/json"},
data=json.dumps(target_data)
)
if response.status_code == 200:
print("Data successfully written to Kingdee Cloud")
else:
print("Failed to write data:", response.text)
```
通过上述步骤和示例代码，可以实现将采购入库的数据从源平台成功转换并写入金蝶云星辰V2API接口。这不仅提高了数据处理效率，还确保了不同系统间的数据一致性和准确性。
