python公司內項目對接釘釘審批流程的實現
最近把組內的一個項目對接釘釘審批接口，通過python3.6。
釘釘官方文檔
廢話不多說瞭，上代碼：
import requests
import json
import time
from dingtalk.crypto import DingTalkCrypto
from django.conf import settings
# settings.BASE_DIR
class Crypto(object):
def __init__(self, token):
# 隨便填的字符串
self.token = token
# 自己生成的43位隨機字符串
self.aes_key = settings.DINGDING.get("DINGTALK_AES_TOKEN")
# 釘釘企業ID
self.corp_id = settings.DINGDING.get("CorpId") #
print("corp_id:", self.corp_id)
self.nonce = settings.DINGDING.get("nonce")
self.crypto = DingTalkCrypto(
token=self.nonce,
encoding_aes_key=self.aes_key,
corpid_or_suitekey=self.corp_id
)
def encrypt_success(self):
# 返回加密success
result = self.crypto.encrypt_message(
msg="success",
nonce=self.nonce,
timestamp=int(time.time()*1000)
)
return result
class DING(object):
def __init__(self, approve_process):
self.AgentId = settings.DINGDING.get("AgentId")
self.AppKey = settings.DINGDING.get("AppKey")
self.AppSecret = settings.DINGDING.get("AppSecret")
self.dingding_url = settings.DINGDING.get("URL")
self.process_code = settings.DINGDING.get("APPROVE_PROCESS").get(approve_process)['process_code']
self.aes_key = settings.DINGDING.get("DINGTALK_AES_TOKEN")
self.nonce = settings.DINGDING.get("nonce")
def get_token(self):
'''
獲取釘釘的token
:return: 釘釘token
'''
url = self.dingding_url + '/gettoken?appkey={}&appsecret={}'.format(self.AppKey, self.AppSecret)
req = requests.get(url)
req = json.loads(req.text)
return req['access_token']
# def createCallbackDd():
# '''
# 註冊釘釘回調函數
# :return:
# '''
# url = 'https://oapi.dingtalk.com/call_back/register_call_back?access_token=' + self.getToken()
# data = {
# "call_back_tag": ["bpms_task_change", "bpms_instance_change"], #這兩個回調種類是審批的
# "token": TOKEN, #自定義的字符串
# "aes_key": AES_KEY, #自定義的43位字符串，密鑰
# "url": URL #回調地址
# }
# requests.post(url, data=json.dumps(data))
# return ('OK')
def create_process(self, originator_user_id, dept_id, form_component_value_vo, approvers, cc_list, has_cc=0):
'''
創建釘釘審批
approvers為list 元素為釘釘userid cc_list同理
'''
url = self.dingding_url + '/topapi/processinstance/create?access_token=' + self.get_token()
print("form_component_value_vo:", form_component_value_vo)
if has_cc == 0:
data = {
'agent_id': self.AgentId,
'process_code': self.process_code, #工單id
'originator_user_id': originator_user_id,
'dept_id': dept_id, #創建人的釘釘部門id
'form_component_values': str(form_component_value_vo), #釘釘後臺配置的需要填寫的字段,
'approvers': approvers,
'cc_list': cc_list,
'cc_position': 'START_FINISH' # 發起和完成時與抄送
}
else:
data = {
'agent_id': self.AgentId,
'process_code': self.process_code, #工單id
'originator_user_id': originator_user_id, #創建人的釘釘userid
'dept_id': dept_id, #創建人的釘釘部門id
'form_component_values': str(form_component_value_vo), #釘釘後臺配置的需要填寫的字段,
'approvers': approvers,
}
print("dingding_utils:", data)
response = requests.post(url, data=data)
return response.json()
def get_status(self, process_instance_id):
url = self.dingding_url + '/topapi/processinstance/get?access_token=' + self.get_token()
data = {
"process_instance_id": process_instance_id
}
response = requests.post(url, data=data)
return response.json()
def register_callback(self, call_back_url):
# 註冊回調
url = self.dingding_url + '/call_back/register_call_back?access_token=' + self.get_token()
print("self.get_token():", self.get_token())
data = {
"call_back_tag": ['bpms_task_change', 'bpms_instance_change'],
"token": self.nonce,
"aes_key": self.aes_key,
"url": call_back_url,
}
response = requests.post(url, data=json.dumps(data))
return response.json()
def get_callback(self):
url = self.dingding_url + '/call_back/get_call_back?access_token=' + self.get_token()
req = requests.get(url)
req = json.loads(req.text)
return req
def create_process_approver_v2(self, originator_user_id, dept_id, form_component_value_vo, approvers, cc_list):
'''
創建釘釘審批
'''
url = self.dingding_url + '/topapi/processinstance/create?access_token=' + self.get_token()
data = {
'agent_id': self.AgentId,
'process_code': self.process_code,
'originator_user_id': originator_user_id,
'dept_id': dept_id,
'form_component_values': str(form_component_value_vo),
'approvers_v2': json.dumps(approvers)
}
if cc_list:
data['cc_list'] = cc_list
data['cc_position'] = 'FINISH'
response = requests.post(url, data=data)
return response.json()
def create_process_approver_v2_test(self, originator_user_id, dept_id, form_component_value_vo):
'''
創建釘釘審批
'''
url = self.dingding_url + '/topapi/processinstance/create?access_token=' + self.get_token()
data = {
'agent_id': self.AgentId,
'process_code': self.process_code,
'originator_user_id': originator_user_id,
'dept_id': dept_id,
'form_component_values': str(form_component_value_vo),
'approvers_v2': json.dumps([
{
"task_action_type": "NONE",
"user_ids": ["dingding_id"], # 單獨審批人
},
{
"task_action_type": "OR",
"user_ids": ["dingding_id1", "dingding_id2"], # 或簽
},
{
"task_action_type": "AND",
"user_ids": ["dingding_id1", "dingding_id2"], # 會簽
}
])
}
response = requests.post(url, data=data)
return response.json()
if __name__ == "__main__":
import django, os, sys
sys.path.append('xxxxxx') # 項目路徑
os.environ['DJANGO_SETTINGS_MODULE'] = 'xx.settings'
# print("settings.DINGDING", settings.DINGDING)
ding = DING("create_xx")
# print(ding.get_token())
# info = [{'name': '單行輸入框','value': 'testixxxxxxxx'}]
# # print(ding.create_process('11', 11, info))
a = [
{'name': "輸入框1", 'value': "value1"},
{'name': "輸入框2", 'value': "value2"},
]
# print(ding.create_process_test('11', 11, a))
# print(ding.create_process_approver_v2_test('11', 11, a))
# print(ding.create_process_test2())
# print(ding.get_status('xxx'))
print(ding.get_status('xx'))
# # 驗證 回調
# a = ding.get_token()
# print(a)
# c = Crypto(a)
# print(c.encrypt_success())
# 註冊回調
# print(ding.register_callback("http://xxxx.vaiwan.com/xxx"))
# print(ding.get_callback())
說明：
1 Crypto類用於對接釘釘回調用的。一個公司隻有一個corpId，並且一個corpid隻能註冊一個回調地址。我司有公共組註冊好瞭回調。隻要接入公司內的回調即可。所以我實際沒有使用到Crypto。
2 在釘釘管理後臺中創建應用後會有這三個東西：AgentId、AppKey，AppSecret 。在創建釘釘審批流程，可以從審批流程瀏覽器中獲取到APPROVE_PROCESS。別忘啦給這個流程審批接口權限。這些官方文檔有說。
3 配置setting變量：
DINGDING = {
"AgentId": 123,
"AppKey": "xx",
"AppSecret": "xx",
"URL": "https://oapi.dingtalk.com",
"APPROVE_PROCESS": { # process_code
"create_xx": {
"process_code": "abc", # 審批流程的id
},
"DINGTALK_AES_TOKEN": "abc",
"nonce": "abc",
"CorpId": "abc",
}
4 接口形式創建的審批流程，與釘釘管理後臺創建的流程有一些不同：
1 不能在不同的審批環節設置不同的抄送人
2 不能審批流程前後有相同的人，不能自動顯示成 “自動同意”(管理後臺設置成去重後，但是接口指定審批人場景，不支持)
5 其他如：審批內容、或簽，會簽代碼裡都有示例。
到此這篇關於python公司內項目對接釘釘審批流程的實現的文章就介紹到這瞭,更多相關python對接釘釘審批內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！