闲庭信步的钉钉机器人调度多群版

前言：影刀高级课程中的“闲庭信步”的钉钉机器人调度，有人看到课程后进行操作只能实现单群的调度，多群调度不知道如何操作，并且回传消息的时候会回传错群，以下是相应的解决办法

1.对应关系

演示效果

2.这里以两个钉钉群为例

配置文件（config.py）

# 配置文件

# 钉钉应用后台配置
DINGDING_INFO = {
    "appSecret_A":"xxxxx", # ReaTea机器人-A
    "appSecret_B":"xxxxx" # ReaTea机器人-B
}

# 影刀调度系统配置
YINGDAO_INFO={
    # 调用系统名称（红茶777）- 服务钉钉A群的API调度系统配置
    "accessKeyId":"xxxxx",
    "accessKeySecret":"xxxxx",
    # 机器人分组UUID
    "robot_grouping":"xxxxx"
}

# 影刀机器人配置
# 应用UUID（ReaTea机器人-A钉钉群可以调用的机器人）
robotUuid_dict_A={
    "ReaTea机器人-A-1号机":"xxxxxx",
    "ReaTea机器人-A-2号机":"xxxxxx",
    "ReaTea机器人-A-3号机":"xxxxxx"
}

# 影刀机器人配置
# 应用UUID（ReaTea机器人-B钉钉群可以调用的机器人）
robotUuid_dict_B={
    "ReaTea机器人-B-1号机":"xxxxx",
    "ReaTea机器人-B-2号机":"xxxxx",
    "ReaTea机器人-B-3号机":"xxxxx"
}

# 消息回传地址配置
# 根据群会话id匹配对应群机器人的webhook地址
# 会话id可以在钉钉群发过来的消息data中获取，参考：cidJpclu3QABc+TZFUDeU4ovA==
group_address_dict={
    # （ReaTea机器人-A钉钉群）
    "A群会话id":["ReaTea机器人-A","https://oapi.dingtalk.com/robot/send?access_token=d80c9ad4a540c512938a5872194613ac5664e2d757cb0721368b65d80ff41dd2"],
    # （ReaTea机器人-B钉钉群）
    "B群会话id":["ReaTea机器人-B","https://oapi.dingtalk.com/robot/send?access_token=ced2c7c5ac8546c97ce152a9042cb55f0f6c5b4d9794268d949c119cec7fceae"]
}

说明：

钉钉应用后台配置：在钉钉应用开发平台创建两个应用—分别获取appSecret；

 A应用：appSecret_A

 B应用：appSecret_B

影刀调度系统配置：影刀控制台—API配置—点击新增按钮—创建调度系统—获取accessKeyId和  accessKeySecret；

每个群的同一个应用可以分配不同的影刀机器人去执行，所以需要配置机器人分组的UUID；

影刀机器人配置：A,B两个群所调用的影刀机器人不同，需要分开处理（应用名称+UUID）；

 （robotUuid_dict_A）：钉钉群A可调用的机器人存放位置

 （robotUuid_dict_B）：钉钉群A可调用的机器人存放位置

消息回传地址配置：根据群会话id匹配对应群机器人的webhook地址。在群里@机器人钉钉会将消息转发到

服务器（ "conversationId": "cidJpclu3QABc+TZFUDeU4ovA=="）。这里的conversationId就是群会话id，让 它与群机器人的webhook地址绑定即可。

如何保证影刀回传的消息可以正确的回传到指定群中？就是把群会话id，全程携带，从钉钉消息中获取到后，作为输入参数传入影刀，执行完毕后作为输出参数回传，这样就可以保证能够回传到正确的群组中。

3.配置路由地址(main.py)

from flask import Flask,request,jsonify
from dingding import verify_sign_A,verify_sign_B,send_msg_to_dingtalk
from yingdao_A import start_application_A
from yingdao_B import start_application_B
import json

app = Flask(__name__) # 创建web实例

# 创建ReaTea机器人-A web路由
@app.route('/dingtalk-A',methods=["post"])
def handle_receive_message_A():
    data=request.get_json() 
    print(f"接收到的消息内容：\n {data}")
    timesamp =request.headers.get("timestamp") 
    received_sign=request.headers.get("sign") 
    if verify_sign_A(timesamp,received_sign):
        print("签名校验通过")
        start_application_A(data)
        return jsonify(),200
    else:
        print("签名校验失败")
        return jsonify(),403

# 创建ReaTea机器人-B web路由
@app.route('/dingtalk-B',methods=["post"])
def handle_receive_message_B():
    data=request.get_json() 
    print(f"接收到的消息内容：\n {data}")
    timesamp =request.headers.get("timestamp") 
    received_sign=request.headers.get("sign") 
    if verify_sign_B(timesamp,received_sign):
        print("签名校验通过")
        start_application_B(data)
        return jsonify(),200
    else:
        print("签名校验失败")
        return jsonify(),403

4.钉钉校验签名合法性（dingding.py）

import hmac
import hashlib
import base64
from config import DINGDING_INFO,group_address_dict
import time
import json
import requests

# ReaTea机器人-A 签名验证
def verify_sign_A(received_timestamp,received_sign):
    app_secret = DINGDING_INFO["appSecret_A"] # 读取appSecret
    app_secret_enc = app_secret.encode("utf-8") # 设置编码格式
    # 进行A群签名校验
    string_to_sign = "{}\n{}".format(received_timestamp,DINGDING_INFO["appSecret_A"])
    string_to_sign_enc = string_to_sign.encode("utf-8")
    hmac_code = hmac.new(app_secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    sign = base64.b64encode(hmac_code).decode('utf-8')
    print(sign)
    if abs(round(time.time()*1000) -int(received_timestamp)) <=3600*1000 and received_sign ==sign:
        return True
    else:
        return False

# ReaTea机器人-B 签名验证
def verify_sign_B(received_timestamp,received_sign):
    app_secret = DINGDING_INFO["appSecret_B"] # 读取appSecret
    app_secret_enc = app_secret.encode("utf-8") # 设置编码格式
    # 进行B群签名校验
    string_to_sign = "{}\n{}".format(received_timestamp,DINGDING_INFO["appSecret_B"])
    string_to_sign_enc = string_to_sign.encode("utf-8")
    hmac_code = hmac.new(app_secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    sign = base64.b64encode(hmac_code).decode('utf-8')
    print(sign)
    if abs(round(time.time()*1000) -int(received_timestamp)) <=3600*1000 and received_sign ==sign:
        return True
    else:
        return False

5.调用影刀应用(分两个.py文件处理)

yingdao_A.py

 应用config.py中的robotUuid_dict_A机器人组

from config import YINGDAO_INFO,robotUuid_dict_A
import requests
import re

#1.获取token
def get_access_tocken():
    url="https://api.yingdao.com/oapi/token/v2/token/create"
    headers={
        "Content-Type":"application/x-www-form-urlencoded"
    }
    params={
        "accessKeyId":YINGDAO_INFO["accessKeyId"],
        "accessKeySecret":YINGDAO_INFO["accessKeySecret"]
    }
    response = requests.post(url=url,headers=headers,params=params)
    return response.json()['data']['accessToken']

#2、启动应用
def start_application_A(data):
    robotName=""
    text = data['text']['content']
    match = re.search(r'【(.*)】', text) # 设置@群机器人调用应用的名称格式
    if match:
        robotName = match.group(1)
    else:
        print("未提取到机器人名称")
    url="https://api.yingdao.com/oapi/dispatch/v2/job/start"
    headers={
        "Authorization":f"Bearer {get_access_tocken()}",
        "Content-Type":"application/json"
    }
    try:
        body={
            "robotClientGroupUuid": YINGDAO_INFO["robot_grouping"],
            "robotUuid": robotUuid_dict_A[robotName],
            "params":[
                {
                "name":"用户id", 
                "value":data['senderStaffId'],
                "type":"str" 
                },
                {
                "name":"消息内容", 
                "value":data['text']['content'],
                "type":"str" 
                },
                {
                "name":"会话id", 
                "value":data["conversationId"],
                "type":"str" 
                }
            ]
        }
        response = requests.post(url=url,headers=headers,json=body)
        return response.json()['data']['jobUuid']
    except Exception as e:
        print("应用名称错误：",e)

yingdao_B.py

  应用config.py中的robotUuid_dict_B机器人组

from config import YINGDAO_INFO,robotUuid_dict_B
import requests
import re

#1.获取token
def get_access_tocken():
    url="https://api.yingdao.com/oapi/token/v2/token/create"
    headers={
        "Content-Type":"application/x-www-form-urlencoded"
    }
    params={
        "accessKeyId":YINGDAO_INFO["accessKeyId"],
        "accessKeySecret":YINGDAO_INFO["accessKeySecret"]
    }
    response = requests.post(url=url,headers=headers,params=params)
    return response.json()['data']['accessToken']

#get_access_tocken()

#2、启动应用
def start_application_B(data):
    robotName=""
    text = data['text']['content']
    match = re.search(r'【(.*)】', text)
    if match:
        robotName = match.group(1)
    else:
        print("未提取到机器人名称")
    url="https://api.yingdao.com/oapi/dispatch/v2/job/start"
    headers={
        "Authorization":f"Bearer {get_access_tocken()}",
        "Content-Type":"application/json"
    }
    try:
        body={
            "robotClientGroupUuid": YINGDAO_INFO["robot_grouping"],
            "robotUuid": robotUuid_dict_B[robotName],
            "params":[
                {
                "name":"用户id", 
                "value":data['senderStaffId'],
                "type":"str" 
                },
                {
                "name":"消息内容", 
                "value":data['text']['content'],
                "type":"str" 
                },
                {
                "name":"会话id", 
                "value":data["conversationId"],
                "type":"str" 
                }
            ]
        }
        response = requests.post(url=url,headers=headers,json=body)
        return response.json()['data']['jobUuid']
    except Exception as e:
        print("应用名称错误：",e)

6.消息回传到钉钉群

main.py 增加回调路由@app.route('/yingdao',methods=["post"])

 同时在控制台中的API配置中添加回调地址

from flask import Flask,request,jsonify
from dingding import verify_sign_A,verify_sign_B,send_msg_to_dingtalk
from yingdao_A import start_application_A
from yingdao_B import start_application_B
import json

app = Flask(__name__) # 创建web实例

# 创建ReaTea机器人-A web路由
@app.route('/dingtalk-A',methods=["post"])
def handle_receive_message_A():
    data=request.get_json() 
    print(f"接收到的消息内容：\n {data}")
    timesamp =request.headers.get("timestamp") 
    received_sign=request.headers.get("sign") 
    if verify_sign_A(timesamp,received_sign):
        print("签名校验通过")
        start_application_A(data)
        return jsonify(),200
    else:
        print("签名校验失败")
        return jsonify(),403

# 创建ReaTea机器人-B web路由
@app.route('/dingtalk-B',methods=["post"])
def handle_receive_message_B():
    data=request.get_json() 
    print(f"接收到的消息内容：\n {data}")
    timesamp =request.headers.get("timestamp") 
    received_sign=request.headers.get("sign") 
    if verify_sign_B(timesamp,received_sign):
        print("签名校验通过")
        start_application_B(data)
        return jsonify(),200
    else:
        print("签名校验失败")
        return jsonify(),403

# 调度系统公用一个回调接口
@app.route('/yingdao',methods=["post"])
def handle_yingdao_callback():
    data= request.get_json()
    print(f"影刀回传消息：\n{data}")
    send_msg_to_dingtalk(data)
    return jsonify(),200

if __name__=='__main__':
    app.run(host="0.0.0.0",port=9527)

dingding.py

增加send_msg_to_dingtalk()方法，将消息回传到对应钉钉群聊。

import hmac
import hashlib
import base64
from config import DINGDING_INFO,group_address_dict
import time
import json
import requests

# ReaTea机器人-A 签名验证
def verify_sign_A(received_timestamp,received_sign):
    app_secret = DINGDING_INFO["appSecret_A"] # 读取appSecret
    app_secret_enc = app_secret.encode("utf-8") # 设置编码格式
    string_to_sign = "{}\n{}".format(received_timestamp,DINGDING_INFO["appSecret_A"])
    string_to_sign_enc = string_to_sign.encode("utf-8")
    hmac_code = hmac.new(app_secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    sign = base64.b64encode(hmac_code).decode('utf-8')
    print(sign)
    if abs(round(time.time()*1000) -int(received_timestamp)) <=3600*1000 and received_sign ==sign:
        return True
    else:
        return False

# ReaTea机器人-B 签名验证
def verify_sign_B(received_timestamp,received_sign):
    app_secret = DINGDING_INFO["appSecret_B"] # 读取appSecret
    app_secret_enc = app_secret.encode("utf-8") # 设置编码格式
    string_to_sign = "{}\n{}".format(received_timestamp,DINGDING_INFO["appSecret_B"])
    string_to_sign_enc = string_to_sign.encode("utf-8")
    hmac_code = hmac.new(app_secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    sign = base64.b64encode(hmac_code).decode('utf-8')
    print(sign)
    if abs(round(time.time()*1000) -int(received_timestamp)) <=3600*1000 and received_sign ==sign:
        return True
    else:
        return False

# 将消息回传到钉钉群聊
def send_msg_to_dingtalk(data):
    back_data = data["result"][0]["value"]
    print("回传的数据",back_data)
    output = json.loads(back_data)
    conversation_id = output["conversation_id"]
    url = group_address_dict[conversation_id][1]
    data = {
        "msgtype": "markdown",
        "markdown": {
            "title": "应用回传消息",
            "text": f"@{output['user_id']}\n{output['content']}"
        },
        "at": {
            "atUserIds": [
                output['user_id']
            ]
        }
    }
    requests.post(url=url, json=data)

7.影刀应用说明

输入参数：用户id、消息内容、会话id

输出参数：回传内容，用户id，会话id

回传消息格式：{"content":"消息内容","user_id":"机器人回复时@的人id","conversation_id":"会话id"}