旺店通·企业奇门数据集成至金蝶云星空：调拨单案例解析
在处理复杂的跨平台数据集成任务时，保证数据一致性和高效传输是关键。本文将聚焦于一个具体的系统对接集成案例，即将旺店通·企业奇门的数据顺利集成到金蝶云星空——尤以“调拨单”的对接为例，来详细探讨技术实现过程。
1. 数据抓取与接口调用
首先，我们利用旺店通·企业奇门提供的API wdt.stock.transfer.query 从其系统中定时可靠地抓取调拨单数据。为了确保不漏单，我们实现了分页策略，并结合接口限流机制，以应对大规模请求时可能遇到的流控问题。
# 调用界面示例代码（伪代码）
params = {
'page_no': page_number,
'page_size': 100,
}
response = wdt_api_call('wdt.stock.transfer.query', params)
data_list = response['data']2. 数据格式转换与映射
由于旺店通·企业奇门和金蝶云星空之间采用的数据格式存在差异，需要进行适当的数据转换和映射。例如，对日期格式、数值精度等细节都进行了特别处理，以满足目标系统要求。
# 简化版数据映射示例（伪代码）
for item in data_list:
mapped_item = {
'outerTransferOrderNo': item['transfer_order_no'],
'quantity': float(item['qty']),
# 更多字段映射...
}
formatted_data.append(mapped_item)3. 批量写入及异常处理
经过清洗和转换后，将结构化的调拨单批量写入到金蝶云星空。我们使用的是其提供的API batchSave 来执行此操作，同时设计了完善的错误重试机制，以确保即使在网络抖动或服务故障情况下，也能自动恢复提交事务，避免重复发送或丢失重要业务信息。
try:
write_response = kingdee_api_call('batchSave', formatted_data)
except Exception as e:
log_error(e, formatted_data)
retry_write_to_kingdee(formatted_data) 通过这种严谨而灵活的方法，不仅可以有效提升处理效率，还能够保障数据的一致性及准确性。在整个过程中，我们不仅实时监控各个环节，还保持详细日志记录，为后续运维工作提供可靠依据。这正是本次集成项目成功实施的重要技术基础部分。
调用旺店通·企业奇门接口获取并加工数据
在数据集成的生命周期中，调用源系统接口是至关重要的一步。本文将详细探讨如何通过轻易云数据集成平台调用旺店通·企业奇门接口 wdt.stock.transfer.query 获取调拨单数据，并进行初步的数据加工。
接口调用配置
首先，我们需要配置元数据，以便正确地调用 wdt.stock.transfer.query 接口。以下是关键的元数据配置：
{
"api": "wdt.stock.transfer.query",
"method": "POST",
"number": "transfer_no",
"id": "transfer_id",
"pagination": {
"pageSize": 100
},
"idCheck": true,
"condition": [
[
{"field":"type","logic":"eqv2","value":"1"},
{"field":"mode","logic":"eqv2","value":"0"},
{"field":"to_warehouse_no","logic":"like","value":"JS"}
],
[
{"field":"type","logic":"eqv2","value":"1"},
{"field":"mode","logic":"eqv2","value":"2"},
{"field":"to_warehouse_no","logic":"like","value":"JS"}
]
],
"request": [
{
"field": "start_time",
"label": "开始时间",
"type": "datetime",
"describe": "增量获取数据，start_time作为开始时间，格式：yyyy-MM-dd HH:mm:ss",
"value": "{{LAST_SYNC_TIME|datetime}}"
},
{
"field": "end_time",
"label": "结束时间",
"type": "datetime",
"describe": "增量获取数据，end_time作为结束时间，格式：yyyy-MM-dd HH:mm:ss",
"value": "{{CURRENT_TIME|datetime}}"
},
{
"field": "from_warehouse_no",
"label": "源仓库",
"type": "string",
"describe":
用于获取指定仓库单据数据信息(不支持一次推送多个仓库编号)"
},
{
...请求参数解析
- start_time 和 end_time：用于增量获取数据，分别表示开始和结束时间。格式为
yyyy-MM-dd HH:mm:ss。 - from_warehouse_no 和 to_warehouse_no：分别代表源仓库和目标仓库的唯一编码，用于区分不同仓库。
- status：调拨单状态，这里设置为
90表示调拨完成。 - type 和 mode：分别表示调拨类型和调拨模式，这里我们设置了两种模式（0和2），均为调拨类型1。
数据请求与清洗
在调用接口后，我们会得到一个包含多个调拨单信息的响应。为了确保数据的准确性和一致性，需要对返回的数据进行清洗和初步处理。
def clean_data(response):
cleaned_data = []
for item in response['data']:
if item['status'] == '90': # 确保只处理已完成的调拨单
cleaned_data.append({
'transfer_id': item['transfer_id'],
'transfer_no': item['transfer_no'],
'from_warehouse': item['from_warehouse_no'],
'to_warehouse': item['to_warehouse_no'],
'status': item['status'],
'created_at': item['created_at']
})
return cleaned_data分页处理
由于接口返回的数据可能非常庞大，我们需要进行分页处理。元数据中的分页配置如下：
{
...
otherRequest: [
{
field: 'page_size',
label: '分页大小',
type: 'string',
describe: '每页返回的数据条数',
value: '{PAGINATION_PAGE_SIZE}'
},
{
field: 'page_no',
label: '页号',
type: 'string',
describe: '不传值默认从0页开始',
value: '{PAGINATION_START_PAGE}'
}
]
}在实际实现中，可以通过循环来逐页获取数据：
def fetch_all_data():
page_no = 0
all_data = []
while True:
response = call_api(page_size=100, page_no=page_no)
if not response['data']:
break
cleaned_data = clean_data(response)
all_data.extend(cleaned_data)
page_no += 1
return all_data数据转换与写入
在完成数据请求与清洗后，我们可以将清洗后的数据转换为目标系统所需的格式，并写入到金蝶JS系统中。这一步骤涉及到具体业务逻辑的实现，如字段映射、格式转换等。
def transform_and_write(data):
transformed_data = []
for item in data:
transformed_item = {
'ID': item['transfer_id'],
'Number': item['transfer_no'],
...
}
transformed_data.append(transformed_item)
write_to_js(transformed_data)通过以上步骤，我们实现了从旺店通·企业奇门接口获取并加工调拨单数据的全过程。这一过程不仅保证了数据的一致性和准确性，还提高了业务处理效率。
数据转换与写入金蝶云星空API接口的技术案例
在数据集成的生命周期中，第二步是将已经集成的源平台数据进行ETL转换，并转为目标平台金蝶云星空API接口所能够接收的格式，最终写入目标平台。本文将详细探讨如何通过轻易云数据集成平台实现这一过程。
配置元数据
首先，我们需要配置元数据，以确保数据能够正确地从源平台（如旺店通）转换并写入到金蝶云星空。以下是我们使用的元数据配置：
{
"api": "batchSave",
"method": "POST",
"idCheck": true,
"operation": {
"rowsKey": "array",
"rows": 20,
"method": "batchArraySave"
},
"request": [
{"field":"FBillNo","label":"单据编号","type":"string","describe":"单据编号","value":"{transfer_no}"},
{"field":"FBillTypeID","label":"单据类型","type":"string","describe":"单据类型","value":"_function case when '{from_warehouse_no}' like '%JS%' then 'ZJDB09_SYS' when '{to_warehouse_no}' like '%JS%' then 'ZJDB07_SYS' else 'DPDB' end"},
{"field":"FBizType","label":"业务类型","type":"string","describe":"下拉列表","value":"NORMAL"},
{"field":"FTransferDirect","label":"调拨方向","type":"string","describe":"下拉列表","value":"GENERAL"},
{"field":"FTransferBizType","label":"调拨类型","type":"string","describe":"下拉列表","value":"ZJDB07_SYS"},
{"field":"FSaleOrgId","label":"销售组织","type":"string","describe":"组织"},
{"field":"FSettleOrgId","label":"结算组织","type":"string","describe":"组织"},
{"field":"FStockOutOrgId","label":"调出库存组织","type":"string","describe":"组织", "value": "100"},
{"field": "FOwnerTypeOutIdHead", "label": "调出货主类型", "type": "string", "describe": "多类别基础资料列表", "value": "BD_OwnerOrg"},
{"field": "FOwnerOutIdHead", "label": "调出货主", "type": "string", "describe": "多类别基础资料", "value": "100"},
{"field": "FStockOrgId", "label": "调入库存组织", "type": "string", describe: 组织, value: 100},
{"field": FSETTLECURRID, label: 结算币别, type: string, describe: 基础资料},
{"field": FDate, label: 日期, type: string, describe: 日期, value: {modified}},
{"field": FNote, label: 备注, type: string, describe: 多行文本},
{"field":{"FBillEntry":{"children":[{"FMaterialId":{"parser":{"name"："ConvertObjectParser"，"params"："FNumber"},"parent"："FBillEntry"}},{"FQty":{"parent"："FBillEntry"}},{"FSrcStockId":{"parent"："FBillEntry"}},{"FDestStockId":{"parent"："FBillEntry"}},{"FNoteEntry":{"parent"："FBillEntry"}},{"FISFREE":{"parent"："FBillEntry"}},{"FDestStockStatusId":{"parser":{"name"："ConvertObjectParser"，"params"："FNumber"}}}]}}}
],
otherRequest:[
{ field:"FormId", label:"业务对象表单Id", type:"string", describe:"必须填写金蝶的表单ID如:PUR_PurchaseOrder", value:"STK_TransferDirect"},
{ field:"IsAutoSubmitAndAudit", label:"提交并审核", type:"bool", value:true},
{ field:"IsVerifyBaseDataField", label:"验证基础资料", type:"bool", describe:"是否验证所有的基础资料有效性，布尔类，默认false（非必录）"，value:false},
{ field:"Operation"，label:"执行的操作"，type:"string"，value:"Save"}
]
}数据转换
在数据转换过程中，我们需要根据源平台的数据格式，将其转换为金蝶云星空API能够接受的格式。以下是关键字段及其对应关系：
FBillNo：单据编号，对应于源平台的transfer_no。FBillTypeID：单据类型，根据from_warehouse_no和to_warehouse_no进行条件判断。FSaleOrgId、FSettleOrgId、FStockOutOrgId等字段都需要通过特定解析器进行转换，以匹配金蝶系统中的编码。
写入目标平台
完成数据转换后，我们使用配置好的API接口将数据写入金蝶云星空。具体步骤如下：
- 构建请求体：根据元数据配置，构建符合金蝶云星空API要求的请求体。
- 发送请求：通过HTTP POST方法，将构建好的请求体发送至金蝶云星空API接口。
- 处理响应：接收并处理API返回的响应信息，确保数据成功写入，并根据需要进行错误处理和日志记录。
以下是一个示例代码片段，用于发送请求：
import requests
import json
url = 'https://api.kingdee.com/batchSave'
headers = {'Content-Type': 'application/json'}
data = {
# 根据元数据配置构建的数据
}
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. Status code: {response.status_code}, Response: {response.text}")通过上述步骤，我们可以实现从旺店通到金蝶云星空的数据无缝对接，有效地完成了ETL过程中的转换与写入工作。这不仅提高了业务流程的自动化程度，也确保了数据的一致性和准确性。
