一、准备工作
1、安装阿里云支持包，点击跳转：https://open.dingtalk.com/document/resourcedownload/download-server-sdk
2、注册钉钉开发者账号，点击链接：https://open.dingtalk.com/
3、获取AK,SK
4、USERID通过企业管理后台可以查看每个用户的ID或者通过接口获取
5、PROCESS_CODE 在流程编辑页面的URL上面（是一个奇葩的设定，无语了家人们）
二、开发流程
1、获取企业token
2、提交审批
3、获取审批返回
import os
import sys
from typing import List
from alibabacloud_dingtalk.workflow_1_0.client import Client as dingtalkworkflow_1_0Client
from alibabacloud_dingtalk.oauth2_1_0.client import Client as dingtalkoauth2_1_0Client
from alibabacloud_dingtalk.oauth2_1_0 import models as dingtalkoauth_2__1__0_models
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_dingtalk.workflow_1_0 import models as dingtalkworkflow__1__0_models
from alibabacloud_tea_util import models as util_models
USER_ID = "1405122143xxxxxx"
PROCESS_CODE = "PROC-B897CFC7-CA66-449C-8C70-xxxxxx"
class Sample:
def __init__(self):
pass
@staticmethod
def create_client() -> dingtalkoauth2_1_0Client:
"""
使用 Token 初始化账号Client
@return: Client
@throws Exception
"""
config = open_api_models.Config()
config.protocol = 'https'
config.region_id = 'central'
return dingtalkoauth2_1_0Client(config)
@staticmethod
def create_client_1_0() -> dingtalkworkflow_1_0Client:
"""
使用 Token 初始化账号Client
@return: Client
@throws Exception
"""
config = open_api_models.Config()
config.protocol = 'https'
config.region_id = 'central'
return dingtalkworkflow_1_0Client(config)
@staticmethod
def get_token() -> str:
"""
获取实例TOKEN
"""
client = Sample.create_client()
get_access_token_request = dingtalkoauth_2__1__0_models.GetAccessTokenRequest(
app_key='xxxxx',
app_secret='yyyyyyyyyyyyyyyyy'
)
res = client.get_access_token(get_access_token_request)
accessToken = res.body.access_token
print("accessToken", accessToken)
return accessToken
@staticmethod
def get_user_info(_token, _code):
"""
没用，测试中
"""
client = Sample.create_client()
get_sso_user_info_headers = dingtalkoauth_2__1__0_models.GetSsoUserInfoHeaders()
get_sso_user_info_headers.x_acs_dingtalk_access_token = _token
get_sso_user_info_request = dingtalkoauth_2__1__0_models.GetSsoUserInfoRequest(
code=_code
)
print(_token, _code)
res = client.get_sso_user_info_with_options(get_sso_user_info_request, get_sso_user_info_headers,
util_models.RuntimeOptions())
print("res", res)
@staticmethod
def get_process_schema_info(token, process_code):
"""
获取审批结构，例如审批中有哪些字段需要填写，可获取如下结构
https://open-dev.dingtalk.com/apiExplorer?spm=ding_open_doc.document.0.0.61e032c7Lzbckc#/?devType=org&api=workflow_1.0%23QuerySchemaByProcessCode
_formList = [
{'id': 'TextField_1B9X1OGLRIDC0', 'name': '部门', 'value': "平台"},
{'id': 'MoneyField_265MJ5CX7OLC', 'name': '金额（元）', 'value': "100"},
{'id': 'TextareaField_1JQHWE0CZJ6O0', 'name': '分润明细', 'value': "https://www.baidu.com"},
{'id': 'TextField_12VLX2OCTGNK0', 'name': '对公-公司全称', 'value': "途强"},
{'id': 'TextField_23VZA2Q6HHLS0', 'name': '开户行', 'value': "途强"},
{'id': 'NumberField_5DTYPD4WMP40', 'name': '对公账号', 'value': "123123123"},
{'id': 'TextField_15B01GLYJL280', 'name': '联系人', 'value': "途强"},
{'id': 'TextField_UTT9HNYLMV40', 'name': '联系电话', 'value': "途强"},
{'id': 'TextareaField_BIO57GH5LFC0', 'name': '备注', 'value': "https://www.baidu.com"}
]
"""
client = Sample.create_client_1_0()
query_schema_by_process_code_headers = dingtalkworkflow__1__0_models.QuerySchemaByProcessCodeHeaders()
query_schema_by_process_code_headers.x_acs_dingtalk_access_token = token
query_schema_by_process_code_request = dingtalkworkflow__1__0_models.QuerySchemaByProcessCodeRequest(
process_code=process_code
)
res = client.query_schema_by_process_code_with_options(query_schema_by_process_code_request, query_schema_by_process_code_headers, util_models.RuntimeOptions())
# print("res", res.body.result.schema_content.items)
items = res.body.result.schema_content.items
formComponentObj = {
"name":"",
"value":""
}
formComponentValues = []
for item in items:
Obj = {
"id": item.props.id,
"name": item.props.label,
"value": 123
}
formComponentValues.append(Obj)
return formComponentValues
@staticmethod
def post_process_instance(token, user_id, process_code, form_list):
"""
创建一个新的审批， 每一个字段的name value 是必填字段
https://open-dev.dingtalk.com/apiExplorer?spm=ding_open_doc.document.0.0.61e032c7Lzbckc#/?devType=org&api=workflow_1.0%23StartProcessInstance
_formList = [
{'id': 'TextField_1B9X1OGLRIDC0', 'name': '部门', 'value': "平台"},
{'id': 'MoneyField_265MJ5CX7OLC', 'name': '金额（元）', 'value': "100"},
...
]
"""
client = Sample.create_client_1_0()
start_process_instance_headers = dingtalkworkflow__1__0_models.StartProcessInstanceHeaders()
start_process_instance_headers.x_acs_dingtalk_access_token = token
form_component_values = []
for item in form_list:
values_obj = dingtalkworkflow__1__0_models.StartProcessInstanceRequestFormComponentValues(
id=item['id'],
name=item['name'],
value=item['value']
)
form_component_values.append(values_obj)
# break
start_process_instance_request = dingtalkworkflow__1__0_models.StartProcessInstanceRequest(
dept_id=890417386,
originator_user_id=user_id,
process_code=process_code,
form_component_values=form_component_values
)
res = client.start_process_instance_with_options(start_process_instance_request, start_process_instance_headers,
util_models.RuntimeOptions())
print("res====", res)
return res.body.instance_id
@staticmethod
def get_instance_info(token, instance_id):
"""
查询实例审批状态
https://open-dev.dingtalk.com/apiExplorer?spm=ding_open_doc.document.0.0.61e032c7Lzbckc#/?devType=org&api=workflow_1.0%23GetProcessInstance
{
"business_id": res.body.result.business_id, # 审批实例业务编号。 可通过钉钉查询
"status": res.body.result.status, # 状态 RUNNING：审批中 TERMINATED：已撤销 COMPLETED：审批完成
"result": res.body.result.result # agree：同意 refuse：拒绝 status为COMPLETED且result为agree时，表示审批单完结并审批通过。
}
"""
client = Sample.create_client_1_0()
get_process_instance_headers = dingtalkworkflow__1__0_models.GetProcessInstanceHeaders()
get_process_instance_headers.x_acs_dingtalk_access_token = token
get_process_instance_request = dingtalkworkflow__1__0_models.GetProcessInstanceRequest(
process_instance_id=instance_id
)
res = client.get_process_instance_with_options(get_process_instance_request, get_process_instance_headers,
util_models.RuntimeOptions())
if res.body.success:
return {
"business_id": res.body.result.business_id, # 审批实例业务编号。 可通过钉钉查询
"status": res.body.result.status, # 状态 RUNNING：审批中 TERMINATED：已撤销 COMPLETED：审批完成
"result": res.body.result.result # agree：同意 refuse：拒绝 status为COMPLETED且result为agree时，表示审批单完结并审批通过。
}
else:
return {}
if __name__ == '__main__':
token = Sample.get_token()
formList = Sample.get_process_schema_info(token, PROCESS_CODE)
# 如果已知审批结构可以直接写死，不用去获取结构
_formList = [
{'id': 'TextField_1B9X1OGLRIDC0', 'name': '部门', 'value': "平台"},
{'id': 'MoneyField_265MJ5CX7OLC', 'name': '金额（元）', 'value': "100"},
{'id': 'TextareaField_1JQHWE0CZJ6O0', 'name': '分润明细', 'value': "https://www.baidu.com"},
{'id': 'TextField_12VLX2OCTGNK0', 'name': '对公-公司全称', 'value': "途强"},
{'id': 'TextField_23VZA2Q6HHLS0', 'name': '开户行', 'value': "途强"},
{'id': 'NumberField_5DTYPD4WMP40', 'name': '对公账号', 'value': "123123123"},
{'id': 'TextField_15B01GLYJL280', 'name': '联系人', 'value': "途强"},
{'id': 'TextField_UTT9HNYLMV40', 'name': '联系电话', 'value': "途强"},
{'id': 'TextareaField_BIO57GH5LFC0', 'name': '备注', 'value': "https://www.baidu.com"}
]
# 提交审批，并返回审批ID
instanceId = Sample.post_process_instance("beab04e9490a3700a577a4c1c15d9698", USER_ID, PROCESS_CODE, _formList)
# 获取审批其他详细信息
result = Sample.get_instance_info("beab04e9490a3700a577a4c1c15d9698", "tlB2uVSURhOhp9SzVPbi3g08441723284181")
# print("result", result)