FASTBI对接金蝶云星空（多组织）
一、基础参数准备
操作说明：
1、使用系统管理员（Administrator）账号密码登录云星空。
2、打开【基础管理】-【公共设置】-【Web API】。
3、在左侧列表中打开任意一张表单，点击【在线测试Web API】，即可找到账套ID。
账套ID
代码：acc_id
1、打开【系统管理】-【系统管理】-【第三方系统登录授权】。
2、在【第三方系统登录授权】页面中，点击【新增】按钮。
3、在【第三方系统登录授权-新增】页面，点击【获取应用ID】按钮。
用户名
代码：user_name
应用ID
代码：app_id
应用秘钥
代码：app_sec
组织编号
代码：org_num
私有云地址（公有云无此参数）
代码：server_url
二、对接API与代码示例
API中心文档：https://openapi.open.kingdee.com/ApiDoc
本次以【凭证表】为例
实现效果：
获取凭证表的部分字段，并日期大于等于2023-03-14小于等于2023-03-15，最后转成表格形式输出
（一）不提取公有参数
Python代码部分
:::info
1、下图几个变量需要根据实际对接的云星辰变化，查找位置请看上述“一、基础参数准备”部分
2、下图“GL_VOUCHER”为对接的接口数据表，可根据实际情况变化，表名见API文档；（此行后部分还有个Limit参数，代表数据返回行数，默认2000，最大10000行）
3、下图为指定接口返回的字段与别名，左侧为接口返回字段，右侧为别名
4、下图为指定过滤条件。
Left=左括号 FieldName=过滤的字段为fdate Compare=大于等于 Value=具体日期 ……
可以理解为在给指定表加where条件，where (fdate>='2023-03-14')
注意：代码中如果增加了过滤条件，数据整合任务时的部分更新也需要同样的条件，保证数据的一致性。
:::
以下为完整代码：
#!/usr/bin/python
# -*- coding:GBK -*-
import json
import logging
from k3cloud_webapi_sdk.main import K3CloudApiSdk
import time
import unittest
def get_data(config,rows):
# 首先构造一个SDK实例
api_sdk = K3CloudApiSdk()
# 然后初始化SDK，需指定相关参数，否则会导致SDK初始化失败而无法使用：
# 初始化方案一：Init初始化方法，使用conf.ini配置文件
# config_path:配置文件的相对或绝对路径，建议使用绝对路径
# config_node:配置文件中的节点名称
# api_sdk.Init(config_path='../conf.ini', config_node='config')
# 初始化方案二（新增）：InitConfig初始化方法，直接传参，不使用配置文件
# acct_id:第三方系统登录授权的账套ID,user_name:第三方系统登录授权的用户,app_id:第三方系统登录授权的应用ID,app_sec:第三方系统登录授权的应用密钥
# server_url:k3cloud环境url(仅私有云环境需要传递),lcid:账套语系(默认2052),org_num:组织编码(启用多组织时配置对应的组织编码才有效)
# api_sdk.InitConfig('62e25034af8811', 'Administrator', '231784_3d9r4dHJ5OgZ4aUJwe6rTxSMVjTdWooF', 'aae9d547ffde46fe9236fdea40472854')
# #第三方系统登录授权的账套ID
# acct_id = '1612392074101927936'
# #第三方系统登录授权的用户
# user_name = '季亮'
# #第三方系统登录授权的应用ID
# app_id = '243996_Q28D0duE7lke7XyK36Qqy79M0s5+WMtO'
# #第三方系统登录授权的应用密钥
# app_sec = '375747d3c4214b85884586549cf47f2d'
# #启用多组织时配置对应的组织编码才有效
# # org_num = 100
# #k3cloud环境url(仅私有云环境需要传递)
# # server_url = ''
acct_id = config['acct_id']
user_name = config['user_name']
app_id = config['app_id']
app_sec = config['app_sec']
org_num = config['org_num']
server_url = config['server_url']
if org_num is None and server_url is None:
api_sdk.InitConfig(acct_id, user_name, app_id, app_sec)
if org_num is None and server_url is not None:
api_sdk.InitConfig(acct_id=acct_id, user_name=user_name, app_id=app_id, app_sec=app_sec,server_url=server_url)
if org_num is not None and server_url is None:
api_sdk.InitConfig(acct_id=acct_id, user_name=user_name, app_id=app_id, app_sec=app_sec,org_num=org_num)
if org_num is not None and server_url is not None:
api_sdk.InitConfig(acct_id=acct_id, user_name=user_name, app_id=app_id, app_sec=app_sec,org_num=org_num,server_url=server_url)
params = {
"CreateOrgId":0,
"Id":"",
"IsSortBySeq":"false",
"Number":"1"
}
# formId = 'GL_VOUCHER'
# response = api_sdk.View(formId, params)
# result = json.loads(response.text)
#接口返回字段与别名
field_dump = {
"faccountbookid" : "账簿",
"faccbookorgid" : "核算组织",
"fvouchergroupno" : "凭证号",
"fyear" : "年度",
"fperiod" : "期间",
"fdate" : "日期",
"fvouchergroupid" : "凭证字",
"faccountid" : "科目编码",
"faccountname" : "科目名称",
"fdebit" : "借方金额",
"fcredit" : "贷方金额",
}
Field_name = []
Field = []
for i,j in field_dump.items():
Field.append(i)
Field_name.append(j)
FieldKeys = ','.join(Field)
rec = []
###做过滤条件的
FilterString = [{"Left":"(","FieldName":"fdate","Compare":">=","Value":"2023-03-14","Right":")","Logic":"AND"},{"Left":"(","FieldName":"fdate","Compare":"<=","Value":"2023-03-15","Right":")","Logic":""}]
##指定接口数据表
para = {"FormId":"GL_VOUCHER","FieldKeys": FieldKeys,"FilterString":FilterString,"OrderString":"","TopRowCount":0,"StartRow":0,"Limit":2000,"SubSystemId":""}
# 调用接口
response = api_sdk.ExecuteBillQuery(para)
try:
result = json.loads(response)
for rd in result:
rec.append(dict(zip(Field_name, rd)))
if rows>0 and len(rec)>=rows:
break
except Exception as e:
raise e
return rec
def load_data(rows=-1, params=None, preview=False):
# if 'API_acct_id'not in params or params['API_acct_id'] is None or len(str(params['API_acct_id'])) == 0:
# raise Exception("参数[API_acct_id]未输入值" )
# else:
# acct_id = params['API_acct_id']
# if 'API_user_name'not in params or params['API_user_name'] is None or len(str(params['API_user_name'])) == 0:
# raise Exception("参数[API_user_name]未输入值" )
# else:
# user_name = params['API_user_name']
# if 'API_app_id'not in params or params['API_app_id'] is None or len(str(params['API_app_id'])) == 0:
# raise Exception("参数[API_app_id]未输入值" )
# else:
# app_id = params['API_app_id']
# if 'API_app_sec'not in params or params['API_app_sec'] is None or len(str(params['API_app_sec'])) == 0:
# raise Exception("参数[API_app_sec]未输入值" )
# else:
# app_sec = params['API_app_sec']
# org_num = None
# if 'API_org_num' in params and len(str(params['API_org_num'])) > 0:
# org_num = params['API_org_num']
# server_url = None
# if 'API_server_url' in params and len(str(params['API_server_url'])) > 0:
# server_url = params['API_server_url']
#第三方系统登录授权的账套ID
acct_id = 'xxxx'
#第三方系统登录授权的用户
user_name = 'xxx'
#第三方系统登录授权的应用ID
app_id = 'xxxx'
#第三方系统登录授权的应用密钥
app_sec = 'xxxx'
#启用多组织时配置对应的组织编码才有效
org_num = None
#k3cloud环境url(仅私有云环境需要传递)
server_url = None
config = {
"acct_id" : acct_id,
"user_name" : user_name,
"app_id" : app_id,
"app_sec" : app_sec,
"org_num" : org_num,
"server_url" : server_url,
}
rest_data=get_data(config,rows)
return rest_data
if __name__=="__main__":
params = {'API_年月' : None }
load_data(rows=-1, params=params, preview=False)
最后不要忘记保存为.py文件
（二）提取接口公有参数
Python代码部分
:::success
1、“一、基础参数准备”部分的参数都准备好；
2、下图双引号内的内容分别在FASTBI参数管理模块创建参数，类型为单值，默认值为上步骤得到的参数结果值。
对照关系如下：
API_acc_id=账套ID API_user_name=用户名 API_app_id=应用ID API_app_sec=应用秘钥
API_org_num=组织编号 API_server_url=私有云地址
3、下图“GL_VOUCHER”为对接的接口数据表，可根据实际情况变化，表名见API文档；（此行后部分还有个Limit参数，代表数据返回行数，默认2000，最大10000行）
4、下图为指定接口返回的字段与别名，左侧为接口返回字段，右侧为别名
5、下图为指定过滤条件。
Left=左括号 FieldName=过滤的字段为fdate Compare=大于等于 Value=具体日期 ……
可以理解为在给指定表加where条件，where (fdate>='2023-03-14')
注意：代码中如果增加了过滤条件，数据整合任务时的部分更新也需要同样的条件，保证数据的一致性。
:::
#!/usr/bin/python
# -*- coding:GBK -*-
import json
import logging
from k3cloud_webapi_sdk.main import K3CloudApiSdk
import time
import unittest
param_list = ["API_acct_id","API_user_name","API_app_id","API_app_sec","API_org_num","API_server_url"]
def get_data(config):
# 首先构造一个SDK实例
api_sdk = K3CloudApiSdk()
# 然后初始化SDK，需指定相关参数，否则会导致SDK初始化失败而无法使用：
# 初始化方案一：Init初始化方法，使用conf.ini配置文件
# config_path:配置文件的相对或绝对路径，建议使用绝对路径
# config_node:配置文件中的节点名称
# api_sdk.Init(config_path='../conf.ini', config_node='config')
# 初始化方案二（新增）：InitConfig初始化方法，直接传参，不使用配置文件
# acct_id:第三方系统登录授权的账套ID,user_name:第三方系统登录授权的用户,app_id:第三方系统登录授权的应用ID,app_sec:第三方系统登录授权的应用密钥
# server_url:k3cloud环境url(仅私有云环境需要传递),lcid:账套语系(默认2052),org_num:组织编码(启用多组织时配置对应的组织编码才有效)
# api_sdk.InitConfig('62e25034af8811', 'Administrator', '231784_3d9r4dHJ5OgZ4aUJwe6rTxSMVjTdWooF', 'aae9d547ffde46fe9236fdea40472854')
# #第三方系统登录授权的账套ID
# acct_id = 'xxx'
# #第三方系统登录授权的用户
# user_name = 'xx'
# #第三方系统登录授权的应用ID
# app_id = 'xxx'
# #第三方系统登录授权的应用密钥
# app_sec = 'xxxx'
# #启用多组织时配置对应的组织编码才有效
# # org_num = ''
# #k3cloud环境url(仅私有云环境需要传递)
# # server_url = ''
acct_id = config['acct_id']
user_name = config['user_name']
app_id = config['app_id']
app_sec = config['app_sec']
org_num = config['org_num']
server_url = config['server_url']
if org_num is None and server_url is None:
api_sdk.InitConfig(acct_id, user_name, app_id, app_sec)
if org_num is None and server_url is not None:
api_sdk.InitConfig(acct_id=acct_id, user_name=user_name, app_id=app_id, app_sec=app_sec,server_url=server_url)
if org_num is not None and server_url is None:
api_sdk.InitConfig(acct_id=acct_id, user_name=user_name, app_id=app_id, app_sec=app_sec,org_num=org_num)
if org_num is not None and server_url is not None:
api_sdk.InitConfig(acct_id=acct_id, user_name=user_name, app_id=app_id, app_sec=app_sec,org_num=org_num,server_url=server_url)
params = {
"CreateOrgId":0,
"Id":"",
"IsSortBySeq":"false",
"Number":"1"
}
# formId = 'GL_VOUCHER'
# response = api_sdk.View(formId, params)
# result = json.loads(response.text)
field_dump = {
"faccountbookid" : "账簿",
"faccbookorgid" : "核算组织",
"fvouchergroupno" : "凭证号",
"fyear" : "年度",
"fperiod" : "期间",
"fdate" : "日期",
"fvouchergroupid" : "凭证字",
"faccountid" : "科目编码",
"faccountname" : "科目名称",
"fdebit" : "借方金额",
"fcredit" : "贷方金额",
}
Field_name = []
Field = []
for i,j in field_dump.items():
Field.append(i)
Field_name.append(j)
FieldKeys = ','.join(Field)
rec = []
FilterString = [{"Left":"(","FieldName":"fdate","Compare":">=","Value":"2023-03-14","Right":")","Logic":"AND"},{"Left":"(","FieldName":"fdate","Compare":"<=","Value":"2023-03-15","Right":")","Logic":""}]
para = {"FormId":"GL_VOUCHER","FieldKeys": FieldKeys,"FilterString":FilterString,"OrderString":"","TopRowCount":0,"StartRow":0,"Limit":2000,"SubSystemId":""}
# 调用接口
response = api_sdk.ExecuteBillQuery(para)
try:
result = json.loads(response)
for rd in result:
rec.append(dict(zip(Field_name, rd)))
except Exception as e:
raise e
return rec
def load_data(rows=-1, params=None, preview=False):
if 'API_acct_id'not in params or params['API_acct_id'] is None or len(str(params['API_acct_id'])) == 0:
raise Exception("参数[API_acct_id]未输入值" )
else:
acct_id = params['API_acct_id']
if 'API_user_name'not in params or params['API_user_name'] is None or len(str(params['API_user_name'])) == 0:
raise Exception("参数[API_user_name]未输入值" )
else:
user_name = params['API_user_name']
if 'API_app_id'not in params or params['API_app_id'] is None or len(str(params['API_app_id'])) == 0:
raise Exception("参数[API_app_id]未输入值" )
else:
app_id = params['API_app_id']
if 'API_app_sec'not in params or params['API_app_sec'] is None or len(str(params['API_app_sec'])) == 0:
raise Exception("参数[API_app_sec]未输入值" )
else:
app_sec = params['API_app_sec']
org_num = None
if 'API_org_num' in params and len(str(params['API_org_num'])) > 0:
org_num = params['API_org_num']
server_url = None
if 'API_server_url' in params and len(str(params['API_server_url'])) > 0:
server_url = params['API_server_url']
config = {
"acct_id" : acct_id,
"user_name" : user_name,
"app_id" : app_id,
"app_sec" : app_sec,
"org_num" : org_num,
"server_url" : server_url,
}
rest_data=get_data(config)
return rest_data
if __name__=="__main__":
params = {'API_年月' : None }
load_data(rows=-1, params=params, preview=False)
三、FASTBI应用
1、数据整合任务
（1）ETL整合任务-添加任务-自定义建仓任务
（2）基础：自定义任务名称、选择数据仓库、任务执行方式
（3）取数配置：上传code文件（.py后缀的文件）
（3）取数配置：上传code文件（.py后缀的文件）
（4）建仓规则：仓库表名、刷新取数配置、数据更新方式（第一次完全更新，后续可根据数据获取时间范围换成部分更新）
（5）点击确定，之后选择手动更新，完成后查看日志否成功写入数据库
2、数据表
（1）打开数据表管理-新建数据表-自定义数据表
（2）上传code文件、选择更新方式、刷新执行结果、命名，最后点击保存
