跳到主要内容

FASTBI对接金蝶云星空（多组织）

一、基础参数准备​

操作说明：

1、使用系统管理员（Administrator）账号密码登录云星空。

image
Preview

2、打开【基础管理】-【公共设置】-【Web API】。

image
Preview

3、在左侧列表中打开任意一张表单，点击【在线测试Web API】，即可找到账套ID。

账套ID​

代码：acc_id

image
Preview

1、打开【系统管理】-【系统管理】-【第三方系统登录授权】。

image
Preview

2、在【第三方系统登录授权】页面中，点击【新增】按钮。

image
Preview

3、在【第三方系统登录授权-新增】页面，点击【获取应用ID】按钮。

用户名​

代码：user_name

image
Preview

应用ID​

代码：app_id

image
Preview

应用秘钥​

代码：app_sec

image
Preview

组织编号​

代码：org_num

image
Preview

私有云地址（公有云无此参数）​

代码：server_url

image
Preview

二、对接API与代码示例​

API中心文档：https://openapi.open.kingdee.com/ApiDoc

本次以【凭证表】为例

image
Preview

实现效果：

获取凭证表的部分字段，并日期大于等于2023-03-14小于等于2023-03-15，最后转成表格形式输出

（一）不提取公有参数​

Python代码部分

信息

1、下图几个变量需要根据实际对接的云星辰变化，查找位置请看上述“一、基础参数准备”部分

image
Preview

2、下图“GL_VOUCHER”为对接的接口数据表，可根据实际情况变化，表名见API文档；（此行后部分还有个Limit参数，代表数据返回行数，默认2000，最大10000行）

image
Preview

3、下图为指定接口返回的字段与别名，左侧为接口返回字段，右侧为别名

image
Preview

4、下图为指定过滤条件。

Left=左括号 FieldName=过滤的字段为fdate Compare=大于等于 Value=具体日期 ……

可以理解为在给指定表加where条件，where (fdate>='2023-03-14')

注意：代码中如果增加了过滤条件，数据整合任务时的部分更新也需要同样的条件，保证数据的一致性。

image
Preview

以下为完整代码：

#!/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=私有云地址

image
Preview

image
Preview

3、下图“GL_VOUCHER”为对接的接口数据表，可根据实际情况变化，表名见API文档；（此行后部分还有个Limit参数，代表数据返回行数，默认2000，最大10000行）

image
Preview

4、下图为指定接口返回的字段与别名，左侧为接口返回字段，右侧为别名

image
Preview

5、下图为指定过滤条件。

Left=左括号 FieldName=过滤的字段为fdate Compare=大于等于 Value=具体日期 ……

可以理解为在给指定表加where条件，where (fdate>='2023-03-14')

注意：代码中如果增加了过滤条件，数据整合任务时的部分更新也需要同样的条件，保证数据的一致性。

image
Preview

#!/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整合任务-添加任务-自定义建仓任务

image
Preview

（2）基础：自定义任务名称、选择数据仓库、任务执行方式

image
Preview

（3）取数配置：上传code文件（.py后缀的文件）

image
Preview

（3）取数配置：上传code文件（.py后缀的文件）

image
Preview

image
Preview

（4）建仓规则：仓库表名、刷新取数配置、数据更新方式（第一次完全更新，后续可根据数据获取时间范围换成部分更新）

image
Preview

（5）点击确定，之后选择手动更新，完成后查看日志否成功写入数据库

image
Preview

2、数据表​

（1）打开数据表管理-新建数据表-自定义数据表

image
Preview

（2）上传code文件、选择更新方式、刷新执行结果、命名，最后点击保存

image
Preview