使用python調用釘釘開放接口，現實給員工單獨發送釘釘通知消息

一. 准備與分析

1. 釘釘開發文檔地址：https://ding-doc.dingtalk.com/doc#/serverapi2/gh60vz

2. 在釘釘上自建應用，並獲取到對應的AGENT_ID, APP_KEY, APP_SECRET

3. 釘釘開發接口需要設置IP白名單才能訪問

4. 要實現本功能，需要對以下接口開放訪問權限：獲取部門列表，獲取部門userid列表，獲取用戶詳情，發送工作通知消息

5. 功能實現思路：

　　1）通過釘釘開放接口，獲取到企業所有員工的電話號碼與員工userId的映射關系

　　2）間接建立起員工電話號碼與發送工作通知消息接口的關系

　　3）待對接的第三方平台上使用員工電話號碼作為登錄賬號

　　　  即員工用電話號碼在第三方平台登錄后，要發送人的釘釘userid就已確定

二. 實現代碼

# -*- coding: utf-8 -*-
# @Time    : 2020/11/22 12:32
# @Author  : chinablue
# @File    : dingtalk_helper.py

import requests
import objectpath

# 在釘釘上創建應用后獲取
AGENT_ID = 123456
APP_KEY = "123456"
APP_SECRET = "123456"

dd_domain = "https://oapi.dingtalk.com"


class DingtalkHelper():

    def __init__(self):
        self.access_token = self.get_token()

    # 獲取token
    def get_token(self):
        url = f"{dd_domain}/gettoken"
        data = {
            "appkey": APP_KEY,
            "appsecret": APP_SECRET,
        }
        res_json = requests.get(url=url, params=data).json()
        return objectpath.Tree(res_json).execute("$.access_token")

    # 獲取部門列表
    def get_depList(self):
        url = f"{dd_domain}/department/list"
        data = {
            "access_token": self.access_token,
        }
        res_json = requests.get(url=url, params=data).json()
        departmentIds_list = list(objectpath.Tree(res_json).execute("$..*[@.name is not null].id"))

        return departmentIds_list

    # 獲取部門用戶userid列表
    def get_memberList(self, depId: str):
        url = f"{dd_domain}/user/getDeptMember"
        data = {
            "access_token": self.access_token,
            "deptId": depId,
        }
        res_json = requests.get(url=url, params=data).json()

        return res_json

    # 獲取用戶詳情
    def get_userInfo(self, userId):
        url = f"{dd_domain}/user/get"
        data = {
            "access_token": self.access_token,
            "userid": userId,
        }
        res_json = requests.get(url=url, params=data).json()

        userIds_list = list(objectpath.Tree(res_json).execute("$..*[@.userid is not null].(mobile, userid)"))
        userInfo = userIds_list[0]

        return {userInfo.get("mobile"): userInfo.get("userid")}

    # 向企業個人發送釘釘通知
    def send_ddMsg(self, userId, build_url, build_result, project_num, build_person, build_time):
        url = f"{dd_domain}/topapi/message/corpconversation/asyncsend_v2?access_token={self.access_token}"
        msg = {
            "msgtype": "oa",
            "oa": {
                "pc_message_url": build_url,
                "head": {
                    "bgcolor": "FFBBBBBB",
                    "text": build_result
                },
                "body": {
                    "title": f"({build_result})Jenkins Pipeline",
                    "form": [
                        {
                            "key": "項目編號:",
                            "value": f"  {project_num}"
                        },
                        {
                            "key": "構建賬號:",
                            "value": f"  {build_person}"
                        },
                        {
                            "key": "構建時間:",
                            "value": f"  {build_time}"
                        },
                        {
                            "key": "構建結果:",
                            "value": f"  {build_result}"
                        },
                    ],
                    "content": "",
                    "image": "@lADOADmaWMzazQKA",
                }
            }
        }
        data = {
            "agent_id": AGENT_ID,
            "userid_list": userId,
            "msg": str(msg),
        }
        # 發送消息
        requests.post(url=url, params=data)

    ### 獲取公司內所有人員的userId
    def getAll_userIds(self):
        deptIds_list = self.get_depList()

        userIds_list = list()

        for deptId in deptIds_list:
            res_json = self.get_memberList(deptId)
            userIds = objectpath.Tree(res_json).execute("$.userIds")
            userIds_list.extend(userIds)

        return userIds_list

    ### 獲取公司內所有人員的mobile和userId的映射關系表
    def getAll_usersInfo(self):
        userIds_list = self.getAll_userIds()

        usersInfo_dict = dict()

        for userId in userIds_list:
            userInfo = self.get_userInfo(userId)
            usersInfo_dict.update(userInfo)

        return usersInfo_dict

注意事項：

1）requests庫和objectpath庫均為python的第三方庫，其中requests庫用於模擬接口請求，objectpath庫用於提取json中的信息

2）示例中的send_ddMsg()方法說明：主要是在一個自研平台上觸發jenkins流水線后，會將jenkins構建結果通知給jenkins構建人