python钉钉机器人自定义回复

Python

大概有这样的需求

  1. 达到xxx条件。比如是到了某个时刻。机器人自动在群里通知，并@相关的人
  2. 比如在群里回复一个关键词、然后期望得到想要的信息，以达到解放双手的目的。这一块儿就需要自己对接钉钉API来实现了。比如这种问答式的

image.png

明白这样的需求后就去 官网 找文档去了。

其实官网文档说的很清楚了。没啥需要补充的。

分析

  • 需要部署到服务器
  • 除了定时还需要自定义回复
  • 不需要性能太高，简单便捷就好。所以在go和Python之间选择了Python

使用Flash启动

主要逻辑代码

1
2
3
4
5
6
7
8
9
10
11
12
13
from flask import Flask
 
# http://flask.pocoo.org/docs/0.12/api/#flask.Flask
app = Flask(__name__)
 
 
@app.route('/HelloWorld')
def hello_world():
    return "Hello World!"
 
if __name__ == "__main__":
    # http://flask.pocoo.org/docs/0.12/quickstart/#a-minimal-application
    app.run(host='0.0.0.0', port='5000')

用python开启flask web服务时，

  • 你只需要本机访问，那ip只要不设置为0.0.0.0就可以，正常访问就好
  • 如果你需要外网访问，ip需要设置为0.0.0.0，此时，在本机上访问需要使用默认的127.0.0.1（也就是你不设置ip时默认的ip）,在外网上访问则需要使用你本机的ip，不要使用0.0.0.0

也可以这样的方式来启动

1
2
3
4
5
6
7
8
9
if __name__ == "__main__":
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.bind(("", 8083))
    server_socket.listen(120)
    while True:
        client_socket, client_address = server_socket.accept()
        handle_client_process = Process(target=handle_client, args=(client_socket,))
        handle_client_process.start()
        client_socket.close()

然后去实现handle_client 就好了。篇幅有限。完整的代码关注公众号 罗尔街 即可获取

架构图

image.png

流程

  • 用户通过django admin来添加消息配置，即关键字与回复内容。
  • 用户通过钉钉企业内部群中@机器人 + 关键字。
  • 企业机器人收到后，由socket监听服务接收，并根据消息类型进行处理后返回。
  • 企业机器人收到返回的消息后，通过内网穿透工具给的外网映射地址进行回复。
  • 企业内部群显示回复的消息，用户看见回复的消息。

实践

针对定时、或者达到xxx条件触发的，在智能群助手里面添加机器人就好了

image.png

选择 自定义机器人
image.png

然后配置其中一种安全方式即可

image.png

得到webhook地址。

也可以先构建一个curl来测试一下

1
2
3
curl 'https://oapi.dingtalk.com/robot/send?access_token=381c2f405e0f906fd556b27cea9f66864120860b5d8b117bb046e10b6599b050&timestamp=1613211530113&secret=SEC2e67120c5e4affa1177ac25fe8dc77ba1c5b49284a9dc7e1888770bc3b76b1fc' \
   -H 'Content-Type: application/json' \
   -d '{"msgtype": "text","text": {"content": "test"}}'

主要逻辑代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
import base64
import hashlib
import hmac
import json
import time
import urllib
import urllib.parse
import requests


def sign():
    timestamp = str(round(time.time() * 1000))
    secret = "SECe5d17dac6060b76c01ea60aec260fe76c6e0644394b932bfffa963342bb630a1"
    secret_enc = secret.encode('utf-8')
    string_to_sign = '{}\n{}'.format(timestamp, secret)
    string_to_sign_enc = string_to_sign.encode('utf-8')
    hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    sign_res = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return timestamp, sign_res


def send_ding_message(text_info):
    webhook = "https://oapi.dingtalk.com/robot/send?access_token=f069339ad1fcb9410d0e96fd947d9a2bf3416451d01dc97e3ef4256c1fdb2b7a"
    header = {
        "Content-Type": "application/json",
        "Charset": "UTF-8"
    }
    text = text_info
    message = {
        "msgtype": "text",
        "text": {
            "content": text
        },
        "at": {
            # "isAtAll": False
            # "atMobiles": [
            #     "18512345678"
            # ]
            "atUserIds": [
                "lbkgv8q"
            ]
        }
    }
    message_json = json.dumps(message)
    timestamp, sign_res = sign()
    webhook += "&timestamp=" + timestamp + "&sign=" + sign_res
    info = requests.post(url=webhook, data=message_json, headers=header)


def usage():
    send_ding_message('test 1')


if __name__ == "__main__":
    usage()

效果
image.png

针对自定义回复消息的

官方链接

按照链接先添加就好了

既然是需要钉钉回调我们服务的 ，所以要校验token、sign的合法性

1
2
3
4
timestamp = request.headers.get('Timestamp')
sign = request.headers.get('Sign')

if sign(timestamp) == sign:

生成规则，官方给了示例

1
2
3
4
5
6
7
8
9
def sign():
    timestamp = str(round(time.time() * 1000))
    secret = "SECe5d17dac6060b76c01ea60aec260fe76c6e0644394b932bfffa963342bb630a1"
    secret_enc = secret.encode('utf-8')
    string_to_sign = '{}\n{}'.format(timestamp, secret)
    string_to_sign_enc = string_to_sign.encode('utf-8')
    hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    sign_res = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return timestamp, sign_res

当然也可以用ip段来做安全的校验

如果是企业内部网络、并非是公网，则需要用内网穿透。官方也给了示例，这里就不再重复了。

一些可能会遇到的坑

  1. 在使用Python的Flask时、报错 “POST / HTTP/1.1” 405 - 可能是没有允许请求本身的方法
    • 解决方法: 添加method
    • @app.route(‘/‘, methods=[‘GET’, ‘POST’])
  2. 钉钉机器人的webhook是固定的、如果是多个群想用同一个机器人、则需要用app_secret来解决

分享到 评论