Python3.7配合Django2.0来调用钉钉(dingding)在线api实时监测员工考勤打卡情况

2022-08-08 10:52:50 浏览数 (1)

    新冠疫情期间,大多数公司为了避免交叉感染都或多或少的采用了远程办公的方式,这显然是一个明智的选择,基本上钉钉(dingding)作为一个远程办公平台来用的话,虽然差强人意,但是奈何市面上没有啥更好的选择,矬子里拔将军,也还是可以凑合用的,不过远程办公有个问题,就是每天需要检查员工的考勤,居家办公虽然灵活,但是大家究竟有没有办公,则是另外一回事,钉钉提供的解决方案就是考勤在线打卡功能,但是检查出勤钉钉在移动端就有点费劲,需要在钉钉app里点击至少5次,还不能实时刷新,pc端的钉钉oa系统做的更烂,还不如移动端来得方便,另外如果你在一家上千人的企业里,这家企业有大大小小几十个部门,你又非常倒霉的担任这家公司的人事主管,每天按部门来出员工考勤报表就不是一件容易事了,所以利用钉钉开放的接口,使用Django自己打造一套实时监控员工考勤的web平台是我们本次的目的。

    项目背景是一家普通科技公司,大概有五个部门,每个部门100人左右

    首先进入钉钉开放平台 :open-dev.dingtalk.com

    在企业内部开发中,选择小程序,新建一个小程序应用,这里其实也还有别的选择,比如h5微应用,主要是小程序兼容性更好一点。

    填写应用的名称、简介、Logo等基本信息这些按下不表,按照要求填写即可,也不必非得填写真实信息,这里有个坑就是一定不要忘了配置安全域名或者ip,安全域名是当我们的检测平台上线的时候部署的域名,应用可以跟指定的域名进行网络通信,如果不配置的话,请求钉钉接口会报403错误。

    另外还有一个坑,也就是钉钉默认开放的接口仅限于基础权限接口

    如果需要考勤或者签到接口的话,还得单独点击申请,这就有点让人看不懂了,那么多接口,全都得靠用鼠标点击开通,不开通就用不了,这个用户体验真是让人非常酸爽,产品设计成这样,钉钉的pm难辞其咎。

    OK,前置准备工作就已经就绪了,现在我们只要根据官方文档来写接口就可以了,选择服务端api文档:https://ding-doc.dingtalk.com/doc#/serverapi2/gh60vz

    钉钉考勤打卡的接口说明是这样的:

该接口用于返回企业内员工的实际打卡记录。比如,企业给一个员工设定的排班是上午9点和下午6点各打一次卡,但是员工在这期间打了多次,该接口会把所有的打卡记录都返回。 如果只要获取打卡结果数据,不需要详情数据,可使用获取打卡结果接口。 请求方式:POST(HTTPS) 请求地址:https://oapi.dingtalk.com/attendance/listRecord?access_token=ACCESS_TOKEN 请求包结构体: { "userIds": ["001","002"], "checkDateFrom": "yyyy-MM-dd hh:mm:ss", "checkDateTo": "yyyy-MM-dd hh:mm:ss", "isI18n":"false" }

    这里每个接口都需要一个access_token用来鉴权,这个token是用id和秘钥通过接口交换回来的,具体在应用详情里可以获取

    这里我们封装成方法

代码语言:javascript复制
import requests

appkey = '你的key'
appsecret = '你的秘钥'

api_url = "https://oapi.dingtalk.com/gettoken?appkey=%s&appsecret=%s"%(appkey,appsecret)

def get_token():
    res = requests.get(api_url)
    if res.status_code == 200:
        str_res = res.text
        token = (json.loads(str_res)).get('access_token')
        return token

    搞定了token,还需要获取您的部门下所有员工的员工id,因为考勤接口参数只能接受员工id,而非部门id

代码语言:javascript复制
#获取用户id
token = get_token()

#获取用户id
res = requests.get('https://oapi.dingtalk.com/user/simplelist?access_token=%s&department_id=235636048' % token)
str_res = res.text
res = json.loads(str_res)
ids = []
id_name = {}
for val in res['userlist']:
    ids.append(val['userid'])
    id_name[val['userid']] = val['name']

n = 50  #大列表中几个数据组成一个小列表
l = ids
biglist = [l[i:i   n] for i in range(0, len(l), n)]

    最后请求考勤接口即可

代码语言:javascript复制
result_all = []

#请求打卡接口
for x in biglist:
    res = requests.post('https://oapi.dingtalk.com/attendance/list?access_token=%s' % token,json ={'access_token':token,'workDateFrom':'%s 00:00:00' % mydate,'workDateTo':'%s 10:00:00' % mydate,'userIdList':str(x),'offset':0,'limit':50})
    str_res = res.text
    #print(str_res)
    res = json.loads(str_res)
    for xy in res['recordresult']:
        result_all.append(xy)

slist = []
for val in result_all:
    if val['timeResult'] == 'NotSigned':
        print(val['timeResult'])
        mydic = {}
        mydic[id_name[val['userId']]] = '未打卡'
        slist.append(mydic)

    完整的后台Django后台接口

代码语言:javascript复制
import requests
from rest_framework.response import Response
from rest_framework.views import APIView
import time

appkey = '你的key'
appsecret = '你的秘钥'

api_url = "https://oapi.dingtalk.com/gettoken?appkey=%s&appsecret=%s"%(appkey,appsecret)

def get_token():
    res = requests.get(api_url)
    if res.status_code == 200:
        str_res = res.text
        token = (json.loads(str_res)).get('access_token')
        return token

class GetQianHandler(APIView):

    def get(self,request):


        mydate = request.GET.get('date',None)

        if mydate == None:

            return Response({'message':'需要传递日期'})

        
        dt_start = '%s 00:00:00' % mydate
        dt_end = '%s 23:59:59' % mydate
        ts_start = int(time.mktime(time.strptime(dt_start, "%Y-%m-%d %H:%M:%S"))) * 1000
        ts_end = int(time.mktime(time.strptime(dt_end, "%Y-%m-%d %H:%M:%S"))) * 1000

        token = get_token()

        #获取用户id
        res = requests.get('https://oapi.dingtalk.com/user/simplelist?access_token=%s&department_id=235636048' % token)
        str_res = res.text
        res = json.loads(str_res)
        ids = []
        id_name = {}
        for val in res['userlist']:
            ids.append(val['userid'])
            id_name[val['userid']] = val['name']

        n = 50  #大列表中几个数据组成一个小列表
        l = ids
        biglist = [l[i:i   n] for i in range(0, len(l), n)]

        result_all = []

        print('workDateFrom %s 00:00:00' % mydate)

        #请求打卡接口
        for x in biglist:
            res = requests.post('https://oapi.dingtalk.com/attendance/list?access_token=%s' % token,json ={'access_token':token,'workDateFrom':'%s 00:00:00' % mydate,'workDateTo':'%s 10:00:00' % mydate,'userIdList':str(x),'offset':0,'limit':50})
            str_res = res.text
            #print(str_res)
            res = json.loads(str_res)
            for xy in res['recordresult']:
                result_all.append(xy)

        slist = []
        for val in result_all:
            if val['timeResult'] == 'NotSigned':
                print(val['timeResult'])
                mydic = {}
                mydic[id_name[val['userId']]] = '未打卡'
                slist.append(mydic)

        return Response(slist)

    这样,就可以愉快的通过线上平台来实时监测部门员工考勤了,效果是这样的:

0 人点赞