python实现钉钉群报警(数仓和数据库数据条数校验)

登录钉钉客户端,创建一个群,把需要收到报警信息的人员都拉到这个群内.然后点击群右上角的"智能群助手"->“添加机器人”->“自定义”,记录该机器人的webhook值

安全设置必须选一个，选择自定义关键词，在这里我定义的是"数仓报警"，后面凡是包含有数仓告警这四个字的都会触发
点击完成之后会生成类似下面的连接

https://oapi.dingtalk.com/robot/send?access_token=8793e3f4bb1f7844d64327e3505778d7ec0baa71bfb5901fcf1514d7xxxxxxx

测试一下是否可以发送报警信息：

curl 'https://oapi.dingtalk.com/robot/send?access_token=8793e3f4bb1f7844d64327e3505778d7ec0baa71bfb5901fcf1514d7xxxxxxx' -H 'Content-Type: application/json' -d '{"msgtype": "text","text": {"content": "数仓报警:数仓数据检验报警"}}'

import requests
import json
url = 'https://oapi.dingtalk.com/robot/send?access_token=8793e3f4bb1f7844d64327e3505778d7ec0baa71bfb5901fcf1514d7xxxxxxx'
headers = {
"Content-Type": "application/json",
"Chartset": "utf-8"
}
#要发送的文本是json格式
request_data = {
#此消息的类型为固定的text
"msgtype": "text",
"text": {
#消息的内容
"content": "数仓报警：python"
},
"at": {
#被@人的手机号
"atMobiles": [156xxxxxxxx]
#控制@所有人
#"isAtAll": True
}
}
#把json转变为字符串格式数据
send_data = json.dumps(request_data)
#这个是发送post请求，请求钉钉接口
response = requests.post(url=url,headers=headers,data=send_data)
#讲求成功后返回的数据
content = response.content.decode()
#打印
print(content)

效果：

修改结构，具体操作：

importsys
importjson
importrequests
 
url = 'https://oapi.dingtalk.com/robot/send?access_token=8793e3f4bb1f7844d64327e3505778d7ec0baa71bfb5901fcf1514d7xxxxxxx'
 
defWriteLogByDing(content):
  headers ={
    "Content-Type": "application/json",
    "Chartset": "utf-8"
  }
  request_data ={
    "msgtype": "text",
    "text": {
      "content": content
    },
    "at": {
      "atMobiles": [],
      "isAtAll": True
    }
  }
  sendData =json.dumps(request_data)
  response =requests.post(url =url,headers =headers,data =sendData)
  content =response.content.decode()
  print(content)
if__name__ =="__main__":
  content =input('请输入想要的信息')
  # content = sys.argv[1]
  WriteLogByDing(content)

# -*- coding: UTF-8 -*-
import requests
import json
url = 'https://oapi.dingtalk.com/robot/send?access_token=8793e3f4bb1f7844d64327e3505778d7ec0baa71bfb5901fcf1514d7xxxxxx'
headers = {
  "Content-Type": "application/json",
  "Chartset": "utf-8"
}
#要发送的文本是json格式
request_data = {
  #发送链接类型的数据
  "msgtype": "link",
  "link": {
    #链接提示
    "text":"数仓报警：机器人是钉钉群的高级扩展功能。群机器人可以将第三方服务的信息聚合到群聊中，实现自动化的信息同步。例如：通过聚合GitHub，GitLab等源码管理服务，实现源码更
新同步；通过聚合Trello，JIRA等项目协调服务，实现项目信息同步。不仅如此，群机器人支持Webhook协议的自定义接入，支持更多可能性，例如：你可将运维报警提醒通过自定义机器人聚合到>钉钉群。",
    #链接标题
    "title": "自定义机器人协议",
    #图片url地址
    "picUrl": "http://p3.so.qhmsg.com/sdr/200_200_/t013d7a21145c708288.jpg",
    #信息的链接跳转
    "messageUrl": "https://open-doc.dingtalk.com/docs/doc.htm?spm=a219a.7629140.0.0.Rqyvqo&treeId=257&articleId=105735&docType=1"
  }
}
#把json转变为字符串格式数据
send_data = json.dumps(request_data)
#这个是发送post请求，请求钉钉接口
response = requests.post(url=url,headers=headers,data=send_data)
#讲求成功后返回的数据
content = response.content.decode()
#打印 
print(content)

效果：

data_check.py

*- coding: UTF-8 -*-
import subprocess, json, sys, os, datetime
import smtplib
from email.header import Header
from email.mime.text import MIMEText
import pymysql
import datetime
import sys
reload(sys)
sys.setdefaultencoding('utf-8')
try:
    import requests
except Exception as e:
    subprocess.getstatusoutput('pip install requests -i http://pypi.douban.com/simple  --trusted-host pypi.douban.com')
def dingtalk_warning(message):
    webhook = "https://oapi.dingtalk.com/robot/send?access_token=0b502034bcaf2f5c31da44f5cd885d4153c291e448b783d5812c81xxxxxxxx"
    headers = {'Content-Type': 'application/json'}
    data = {
        "msgtype": "text",
        "text": {
            "content": message,
        },
    }
    x = requests.post(url=webhook, data=json.dumps(data), headers=headers)
    if x.json()["errcode"] == 0:
        return True
    else:
        return False
def rangeTableCount(table_name,pt,jdbc_count,hive_count):
    if jdbc_count != hive_count:
        subject = table_name +":" + pt+ "分区数据不相等："
    	dingtalk_warning("业务报警：" + 'n' + subject + 'n' + "jdbc条数：" +str(jdbc_count) +'n'+ "hive条数：" +str(hive_count) )
    
if __name__ == '__main__':
    rangeTableCount(sys.argv[1],sys.argv[2],sys.argv[3],sys.argv[4])

调用脚本时需要传入4个参数，依次是：表名，分区，jdbc数据库中数据条目数，hive数仓中数据条目数

其中hive和mysql查询表条数在shell脚本中执行：

##可以根据自己的情况添加条件
jdbc_count_sql="select count(1) from t_address_region;"

hive_count_sql="select count(1) from ods_qn.ods_edu_address_region_knowledge_wh_df where pt = '$start_date' ;"


jdbc_count=$(echo $(mysql -h 127.0.0.1 -P 3306 -u root-p"root" -e "use test; $jdbc_count_sql") | awk '{print $2}')
hive_count=$(echo $(beeline -u jdbc:hive2://127.0.0.1:10000 -n root-p root -e "$hive_count_sql") | awk -F"|" '{print $4}')

效果：

// 消息内容中不包含任何关键词
{
“errcode”:310000,
“errmsg”:“keywords not in content”
}
// timestamp 无效
{
“errcode”:310000,
“errmsg”:“invalid timestamp”
}
// 签名不匹配
{
“errcode”:310000,
“errmsg”:“sign not match”
}
// IP地址不在白名单
{
“errcode”:310000,
“errmsg”:“ip X.X.X.X not in whitelist”
}