为了实现工单系统，又不想自己去结合钉钉的组织架构实现一套审批流，所以采用钉钉的审批结合自己的系统去实现工单。

由钉钉实现审批流，软件实现查询，记录工单等。

一、具体流程

 

 

二、具体的实现方式

获取token的函数

def getToken(self):
    '''
    获取钉钉的token
    :return: 钉钉token
    '''
    url = 'https://oapi.dingtalk.com/gettoken?appkey=XXXXXXXXXXXXXXXXX&appsecret=XXXXXXXXXXXXXXXXXXXXXXXXXX'
    req = requests.get(url)
    req = json.loads(req.text)
    return req['access_token']

1. 注册回调函数

def createCallbackDd(self, request):
    '''
    注册钉钉回调函数
    :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 HttpResponse('OK')

 

2. 创建钉钉审批

注意：不使用python的dingtalk-sdk，这个版本必须要传approvers（审批人）这个参数，导致不能使用后台配置的工单流，直接使用接口或者找到最新版的sdk

def create(self, request):
    '''
    创建钉钉审批
    :param request:
    :return:
    '''
    url = 'https://oapi.dingtalk.com/topapi/processinstance/create?access_token=' + self.getToken()
    form_component_value_vo = [{'name': '姓名', 'value': '测试2'}, {'name': '部门', 'value': '测试2'}, {'name': '["开始时间","结束时间"]', 'value': '["2019-06-10 10:32","2019-06-10 10:36"]'}, {'name': '加班事由', 'value': '测试2'}] #钉钉后台配置的需要填写的字段（这一段由点到传过来实现，这里只是测试使用）
    data = {
        'agent_id': None,
        'process_code': 'XXXXXXXXXXXXXXXX',  #工单id
        'originator_user_id': 'manager2606', #创建人的钉钉userid
        'dept_id': -1,  #创建人的钉钉部门id
        'form_component_values': str(form_component_value_vo), #钉钉后台配置的需要填写的字段
    }
    response = requests.post(url, data=data)
    print(response.text)
    return HttpResponse('成功')

 

3. 回调函数

在您注册事件回调接口的时候，钉钉服务器会向您“注册回调接口”时候上传的url(接收回调的url)推送一条消息，用来测试url的合法性。收到消息后，需要返回经过加密后的字符串“success”的json数据，否则钉钉服务器将认为url不合法。

class Crypto(object):
    def __init__(self):
        # 随便填的字符串
        self.token = TOKEN
        # 自己生成的43位随机字符串
        self.aes_key = AES_KEY
        # 钉钉企业ID
        self.corpid = CORPID
        self.crypto = DingTalkCrypto(
            token=self.token,
            encoding_aes_key=self.aes_key,
            corpid_or_suitekey=self.corpid
        )

def callbackDd(self, request):
    encrypt = request.data.get('encrypt')

    msgSignature = request.GET.get("signature")
    timeStamp = request.GET.get("timestamp")
    nonce = request.GET.get("nonce")

    callback = json.loads(Crypto().crypto.decrypt_encrypt_str(
        encrypt_str=encrypt,
        nonce=nonce,
        timestamp=timeStamp,
        signature=msgSignature
    ))
    print(callback)
    #第一次调用回调函数的验证，关键字是check_url
    if callback['EventType'] == 'check_url':
        result = Crypto().crypto.encrypt_message(
            msg="success",
            nonce="123456",
            timestamp=int(time.time() * 1000)
        )

        return Response(result)
    # bpms_task_change :  审批任务开始，结束，转交
    # bpms_instance_change：审批实例开始，结束
    elif callback['EventType'] == 'bpms_task_change':
        pass  #实现入库
    elif callback['EventType'] == 'bpms_instance_change':
        pass  #实现入库

 

三、钉钉的后台配置

参考钉钉的文档：https://open-doc.dingtalk.com/microapp/serverapi2/tvu5f4

上面的代码实现，我的配置方式是：

 

结尾：后续的数据库设计会在后面的内容中展示。