手把手教你搭建钉钉预警机器人

2023-11-23 10:00:10

摘要

随着生产上线业务的不断增多以及无人值守需求的日益增长，钉钉预警机器人的出现极大的解决了我们的问题。本文从0到1详细介绍讲解下钉钉预警机器人的搭建与自定义，分为机器人搭建，预警内容开发，上线部署三大步

机器人搭建

以下所有的配置开发均来源于钉钉官网文档，喜欢阅读官网或者有更进一步个性化的需求可前往查看，也可以跟随笔者站在初级开发者的角度来一步步搭建

首先是先要有一个群以及你有这个群添加机器人的权限，然后

添加机器人

群设置--智能群助手--添加机器人--自定义，选择添加，然后进行相关设置

设置机器人

首先起一个好听的名字，安全设置那里有三个，必须要至少有一个

自定义关键词：自定义发送内容关键字，发送的内容里含有设置的关键字才会发送
加签：也就是加密，一般这个用的比较多，
IP地址：这个就很好理解了，特定的IP可以发送内容

这里我们选用第二个加签，然后点击添加，这里群会提示你成功添加了一个机器人

我们打开这个机器人把两个数据保存下，分别是Webhook token和加签的密文

https://oapi.dingtalk.com/robot/send?access_token=xxxxxxxxxxxWebhook: https://oapi.dingtalk.com/robot/send?access_token=xxxxxxxxxxx
加签：yyyyyyyyyyyyyyyyyyy

这两个xxxxxx和yyyyy留存备用

预警内容开发

上述的机器人添加好了，相当于一个接收数据的API接口好了，我们现在开始开发预警内容

采用最为简便的python，整体框架代码如下

from datetime import datetime, timedelta
import requests
import hmac
import hashlib
import base64
import urllib.parse
import urllib.request
import json
import time
import os
import sys
import re
def getUrl(secret,access_token):
    timestamp = str(round(time.time() * 1000))
    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 = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    url = "https://oapi.dingtalk.com/robot/send?access_token=%s&sign=%s&timestamp=%s" % (access_token, sign, timestamp)
    return url
def sendToGroup(url,header,data):
    sendData = json.dumps(data)#将字典类型数据转化为json格式
    sendData = sendData.encode("utf-8") # python3的Request要求data为byte类型
    request = urllib.request.Request(url=url, data=sendData, headers=header)
    opener = urllib.request.urlopen(request)
    return opener


secret= ''
access_token=''
data={
 "msgtype": "text",
 "text": {
     "content": "分析任务已完成"
 },
  "at": {
      "atMobiles": ["13216816870"],
      "isAtAll": False
  }
}

header = {
"Content-Type": "application/json",
"Charset": "UTF-8"
}
url=getUrl(secret,access_token)
opener=sendToGroup(url,header,data)

把上面的xxxx和yyyy放到代码里的secret和access_token，运行即可，钉钉机器人就会弹出

而整个内容需要开发的就是代码里data那一部分

我们常见的查询就是经过数据库，所以我们这里也连接数据库查询，加一个sql 函数代码

def exeSqlRes(dbip,dbport,dbacco,dbpwd,dbname,sql):
    conn = connect(host=dbip, port=dbport, user=dbacco, password=dbpwd, database=dbname) 
    cursor = conn.cursor() 
    try:
        cursor.execute(sql)    
        list1 = cursor.fetchall()
        global dataStr1,dataStr2
        dataStr1 = "行内运行日期 ：%s" % (list1[0][0])
        dataStr2 = "公募运行日期 ：%s" % (list1[1][0])
    except:
        print("数据库连接失败\n")
    conn.close()
sql = '''
    SELECT SZ_TO_CHAR_D(max(d_date),'YYYYMMDD') as SYSTEMDATE from datacollect_log 
    union all 
    SELECT SZ_TO_CHAR_D(max(d_date),'YYYYMMDD') as SYSTEMDATE from datacollect_log 
    )
'''
exeSqlRes('xx.xx.xx.xx',3306,'xxxxx','xxxxx','xxxxx',sql)

data的形式钉钉官网介绍的很多，有txt，json，xml等，限于篇幅，这里我们只列一下最常用的MARKDOWN形式

data = {
     "msgtype": "markdown",
     "markdown": {
         "title":"数据监控",
         "text": "#### 当前数据运行日期 @17665011132 \n > - **%s** \n > - **%s** \n > ![screenshot](https://www.icode9.com/i/ll/?i=3c6a5aa682df459ab04d2fe5dc32475c.png?,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA56We6Iq36L-m6JOd5a-6,size_20,color_FFFFFF,t_70,g_se,x_16)\n > ###### %s发布 [查看详情](https://www.dingtalk.com) \n" % (dataStr1,dataStr2,timeStr)
     },
      "at": {
          "atMobiles": [
              "176xxxx1132"
          ],
          "atUserIds": [
              ""
          ],
          "isAtAll": "false"
      }
 }

大家可根据自己的实际需求进行data开发，我这里开发完毕后运行代码弹出

一切配置完成

上线部署

把这个python代码放到crontab里按需调度运行即可

*/10 * * * * /bisz/python/anaconda3/bin/python /bisz/python/DingDingSend.py

码农公寓

相关文章