DINGTALK_CORP_ID = "重点：企业应用ID"
# 钉钉微应用
DINGTALK_APP_KEY = "微应用APP_KEY"
DINGTALK_CORP_SECRET = "微应用SERCRET"
# 钉钉移动应用(扫码登录)，在移动应用接入菜单里配置
DINGTALK_APP_ID = "移动应用APP_ID"
DINGTALK_APP_SECRET = "移动应用APP_SECRET"
# 钉钉aes加密(随机)
DINGTALK_AES_TOKEN = "43位字符串 a-zA-Z0-9中生成"
DINGTALK_TOKEN = "字符串随便填"

<script src="/UploadFiles/2021-04-08/ddLogin.js">

配置扫码后跳转页面(就是第一步goto参数里的url)，钉钉会自动在跳转页面路由后添加code和state参数

拿到code请求后端接口


<script src="/UploadFiles/2021-04-08/jquery.min.js">

后端用appid和appsecret获取access_token


import requests
from dingtalk.client import SecretClient

class GetInfoByCode(object):
  def __init__(self):
    self.url = "https://oapi.dingtalk.com/sns"

  def get_access_token(self):
    resp = requests.get(
      url=self.url + "/gettoken",
      params=dict(appid=DINGTALK_APP_ID, appsecret=DINGTALK_APP_SECRET)
    )
    resp = resp.json()
    return resp["access_token"]


用access_token和code获取用户unionid


def get_persistent_code(self, code, token):
    resp = requests.post(
      url="%s/get_persistent_code" % (self.url, token),
      json=dict(tmp_auth_code=code),
    )
    resp = resp.json()
    return resp["unionid"]


用unionid换取userid


ding_client = SecretClient(corp_id=DINGTALK_CORP_ID,
               corp_secret=DINGTALK_CORP_SECRET)
# 这块提醒下一定要做缓存，由于我之前没做缓存，超过钉钉使用频率而被限制了,修改下上面那句代码
import redis
from dingtalk.storage.kvstorage import KvStorage

# redis做缓存
redis_ding = redis.StrictRedis(REDIS_HOST, REDIS_PORT, db=0)
# 划重点，因为我是第三方web网站，没有使用E应用，所以我的corp_id填的是微应用的id，而不是企业应用ID(corp_id)
ding_client = SecretClient(
  corp_id=DINGTALK_APP_KEY,
  corp_secret=DINGTALK_CORP_SECRET,
  storage=KvStorage(redis_ding))

userid = ding_client.user.get_userid_by_unionid(unionid)


userid换取user详情


user = ding_client.user.get(userid["userid"])


二. 加解密
应用场景: 钉钉回调
事件流程：注册回调事件—>处理事件内容(新增员工、更新部门等),返回钉钉加密json

就是说你调用注册事件API，钉钉会给你需要注册的url发送一个post请求，你得按要求返回success，否则注册失败
1.注册事件

再看看什么是success的加密json数据

如果你没有看懂的话，很巧 我也看不懂…
json数据包含这些参数

时间戳和nonce容易拿，问题在于encrypt和msg_signature的生成

因为钉钉没有提供python的加解密库，所以看到这块我有点懵


钉钉的文档看不太懂…

百度搜了下 发现有人自己封装了这个库，我看了下名字XXXcrypto，crypto这个单词好像在sdk里边见过。


看了sdk目录，果然，虽然钉钉sdk虽然没写关于加解密的内容，但是它其实已经封装了这个库
使用


from dingtalk.crypto import DingTalkCrypto

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


说明
原本是不需要自己封装的。看dingtalk的源码

在开始定义ding_client的时候，只需要填参数就可以，经过我反复的调试，发现因为我开发的是企业内部应用-微应用,而不是E应用，所以这里填的corp_id其实是微应用app_key，参数不正确导致后边调试一直返回文本非success错误，为了不与上边填的参数冲突，所以自己封装Crypto类，填写的corp_id是企业应用id
封装好Crypto这个类，就可以调用啦, 先测试返回给钉钉的结果是否正确


from ... import Crypto

class EventCallBack(Resource):
  def __init__(self):
    self.crypto = Crypto().crypto
    
  def post(self):
    # 返回加密success
    result = self.crypto.encrypt_message(
      msg="success",
      nonce="123456",
      timestamp=int(time.time()*1000)
    )
    return result


事情没这么容易结束…
这个post方法你需要对回调事件作出处理而不是仅仅返回success，所以最终需要这样， 对结果进行处理


 def post(self):
    po = request.json["encrypt"]
    data = request.values
    # 返回加密success
    result = self.crypto.encrypt_message(
      msg="success",
      nonce="123456",
      timestamp=int(time.time()*1000)
    )
    # 获取到事件及其他信息
    callback = self.crypto.decrypt_encrypt_str(
      encrypt_str=po,
      nonce=data["nonce"],
      timestamp=data["timestamp"],
      signature=data["signature"]
    )
    # demjson可以把字符串转为字典，否则取值失败
    # 安装 pip install demjson
    callback = demjson.decode(callback)
    # 判断事件类型，作出相应回调
    if callback["EventType"] == "user_add_org":
      # 添加员工
      ...
    elif callback["EventType"] == "org_dept_create":
      # 添加部门
      ...
    return result


三. 调试


# 钉钉内网穿透工具，主要作用是方便本地调试需要回调域名的接口
# 地址: https://open-doc.dingtalk.com/microapp/debug/ucof2g
可以后台启动


以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。