公司內專案對接釘釘審批流程(python)
阿新 • • 發佈：2020-08-19
最近把組內的一個專案對接釘釘審批介面，通過python3.6。
釘釘官方文件https://ding-doc.dingtalk.com/doc#/serverapi2/ca8r99
廢話不多說了，上程式碼：
1 import requests 2 import json 3 import time 4 from dingtalk.crypto import DingTalkCrypto 5 6 from django.conf import settings 7 # settings.BASE_DIR 8 9 10 class Crypto(object):11 def __init__(self, token): 12 # 隨便填的字串 13 self.token = token 14 # 自己生成的43位隨機字串 15 self.aes_key = settings.DINGDING.get("DINGTALK_AES_TOKEN") 16 # 釘釘企業ID 17 self.corp_id = settings.DINGDING.get("CorpId") # 18 print("corp_id:", self.corp_id)19 self.nonce = settings.DINGDING.get("nonce") 20 self.crypto = DingTalkCrypto( 21 token=self.nonce, 22 encoding_aes_key=self.aes_key, 23 corpid_or_suitekey=self.corp_id 24 ) 25 26 def encrypt_success(self): 27 # 返回加密success28 result = self.crypto.encrypt_message( 29 msg="success", 30 nonce=self.nonce, 31 timestamp=int(time.time()*1000) 32 ) 33 return result 34 35 36 class DING(object): 37 def __init__(self, approve_process): 38 self.AgentId = settings.DINGDING.get("AgentId") 39 self.AppKey = settings.DINGDING.get("AppKey") 40 self.AppSecret = settings.DINGDING.get("AppSecret") 41 self.dingding_url = settings.DINGDING.get("URL") 42 self.process_code = settings.DINGDING.get("APPROVE_PROCESS").get(approve_process)['process_code'] 43 self.aes_key = settings.DINGDING.get("DINGTALK_AES_TOKEN") 44 self.nonce = settings.DINGDING.get("nonce") 45 46 def get_token(self): 47 ''' 48 獲取釘釘的token 49 :return: 釘釘token 50 ''' 51 url = self.dingding_url + '/gettoken?appkey={}&appsecret={}'.format(self.AppKey, self.AppSecret) 52 req = requests.get(url) 53 req = json.loads(req.text) 54 return req['access_token'] 55 56 # def createCallbackDd(): 57 # ''' 58 # 註冊釘釘回撥函式 59 # :return: 60 # ''' 61 # url = 'https://oapi.dingtalk.com/call_back/register_call_back?access_token=' + self.getToken() 62 # data = { 63 # "call_back_tag": ["bpms_task_change", "bpms_instance_change"], #這兩個回撥種類是審批的 64 # "token": TOKEN, #自定義的字串 65 # "aes_key": AES_KEY, #自定義的43位字串，金鑰 66 # "url": URL #回撥地址 67 # } 68 # requests.post(url, data=json.dumps(data)) 69 # return ('OK') 70 71 def create_process(self, originator_user_id, dept_id, form_component_value_vo, approvers, cc_list, has_cc=0): 72 ''' 73 建立釘釘審批 74 approvers為list 元素為釘釘userid cc_list同理 75 ''' 76 url = self.dingding_url + '/topapi/processinstance/create?access_token=' + self.get_token() 77 print("form_component_value_vo:", form_component_value_vo) 78 if has_cc == 0: 79 data = { 80 'agent_id': self.AgentId, 81 'process_code': self.process_code, #工單id 82 'originator_user_id': originator_user_id, 83 'dept_id': dept_id, #建立人的釘釘部門id 84 'form_component_values': str(form_component_value_vo), #釘釘後臺配置的需要填寫的欄位, 85 'approvers': approvers, 86 'cc_list': cc_list, 87 'cc_position': 'START_FINISH' # 發起和完成時與抄送 88 } 89 else: 90 data = { 91 'agent_id': self.AgentId, 92 'process_code': self.process_code, #工單id 93 'originator_user_id': originator_user_id, #建立人的釘釘userid 94 'dept_id': dept_id, #建立人的釘釘部門id 95 'form_component_values': str(form_component_value_vo), #釘釘後臺配置的需要填寫的欄位, 96 'approvers': approvers, 97 } 98 print("dingding_utils:", data) 99 response = requests.post(url, data=data) 100 return response.json() 101 102 def get_status(self, process_instance_id): 103 url = self.dingding_url + '/topapi/processinstance/get?access_token=' + self.get_token() 104 data = { 105 "process_instance_id": process_instance_id 106 } 107 response = requests.post(url, data=data) 108 return response.json() 109 110 def register_callback(self, call_back_url): 111 # 註冊回撥 112 url = self.dingding_url + '/call_back/register_call_back?access_token=' + self.get_token() 113 print("self.get_token():", self.get_token()) 114 data = { 115 "call_back_tag": ['bpms_task_change', 'bpms_instance_change'], 116 "token": self.nonce, 117 "aes_key": self.aes_key, 118 "url": call_back_url, 119 } 120 response = requests.post(url, data=json.dumps(data)) 121 return response.json() 122 123 def get_callback(self): 124 url = self.dingding_url + '/call_back/get_call_back?access_token=' + self.get_token() 125 req = requests.get(url) 126 req = json.loads(req.text) 127 return req 128 129 def create_process_approver_v2(self, originator_user_id, dept_id, form_component_value_vo, approvers, cc_list): 130 ''' 131 建立釘釘審批 132 ''' 133 url = self.dingding_url + '/topapi/processinstance/create?access_token=' + self.get_token() 134 data = { 135 'agent_id': self.AgentId, 136 'process_code': self.process_code, 137 'originator_user_id': originator_user_id, 138 'dept_id': dept_id, 139 'form_component_values': str(form_component_value_vo), 140 'approvers_v2': json.dumps(approvers) 141 } 142 if cc_list: 143 data['cc_list'] = cc_list 144 data['cc_position'] = 'FINISH' 145 146 response = requests.post(url, data=data) 147 return response.json() 148 149 def create_process_approver_v2_test(self, originator_user_id, dept_id, form_component_value_vo): 150 ''' 151 建立釘釘審批 152 ''' 153 url = self.dingding_url + '/topapi/processinstance/create?access_token=' + self.get_token() 154 data = { 155 'agent_id': self.AgentId, 156 'process_code': self.process_code, 157 'originator_user_id': originator_user_id, 158 'dept_id': dept_id, 159 'form_component_values': str(form_component_value_vo), 160 'approvers_v2': json.dumps([ 161 { 162 "task_action_type": "NONE", 163 "user_ids": ["dingding_id"], # 單獨審批人 164 }, 165 { 166 "task_action_type": "OR", 167 "user_ids": ["dingding_id1", "dingding_id2"], # 或籤 168 }, 169 { 170 "task_action_type": "AND", 171 "user_ids": ["dingding_id1", "dingding_id2"], # 會籤 172 } 173 ]) 174 } 175 176 response = requests.post(url, data=data) 177 return response.json() 178 179 180 if __name__ == "__main__": 181 182 import django, os, sys 183 sys.path.append('xxxxxx') # 專案路徑 184 os.environ['DJANGO_SETTINGS_MODULE'] = 'OMS.settings' 185 # print("settings.DINGDING", settings.DINGDING) 186 ding = DING("create_sendexplain") 187 # print(ding.get_token()) 188 # info = [{'name': '單行輸入框','value': 'testixxxxxxxx'}] 189 # # print(ding.create_process('11', 11, info)) 190 a = [ 191 {'name': "輸入框1", 'value': "value1"}, 192 {'name': "輸入框2", 'value': "value2"}, 193 ] 194 # print(ding.create_process_test('11', 11, a)) 195 # print(ding.create_process_approver_v2_test('11', 11, a)) 196 # print(ding.create_process_test2()) 197 # print(ding.get_status('xxx')) 198 print(ding.get_status('xx')) 199 200 # # 驗證 回撥 201 # a = ding.get_token() 202 # print(a) 203 # c = Crypto(a) 204 # print(c.encrypt_success()) 205 206 # 註冊回撥 207 # print(ding.register_callback("http://xxxx.vaiwan.com/xxx")) 208 # print(ding.get_callback())
說明：
1Crypto類用於對接釘釘回撥用的。一個公司只有一個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其他如：審批內容、或籤，會籤程式碼裡都有示例。