“我报名参与金石计划1期应战——瓜分10万奖池，这是我的第1篇文章，点击检查活动概况”

先导

作业中运维作业经常会遇到一些数据报告,数据监控, 作为一个新人诚心感觉这些数据没有什么意思(当然也许是我菜),有句话怎么说,懒散是人类进步的阶梯.这儿运用 python 衔接数据把数据 传到钉钉, 这样可以进行数据监控 ,看看是哪家小可爱又 搞事情了 ╭(╯^╰)╮

钉钉接口

钉钉供给了群机器人接口等许多许多接口,网上 demo 一大堆.这儿只制造简单的表述

钉钉官网:

[developers.dingtalk.com/document/ap…]

可以给群里增加一个机器人，经过 @固定人 ，或许所有人，广播等方法发送信息

创建钉钉机器人

首先你得建立个群

PYTHON 连接钉钉传输工作数据监控

增加群机器人

完结必要的安全设置，勾选我已阅览并赞同《自定义机器人服务及免责条款》，然后单击完结。

现在有 3 种安全设置方法，请根据需要挑选一种：

  • 自定义关键词：最多可以设置 10 个关键词，音讯中至少包括其中 1 个关键词才可以发送成功。

    例如增加了一个自定义关键词：监控报警，则这个机器人所发送的音讯，必须包括监控报警这个词，才能发送成功。

  • 加签：

    把timestamp+"\n"+密钥当做签名字符串，运用 HmacSHA256 算法计算签名，然后进行 Base64 encode，终究再把签名参数再进行 urlEncode，得到终究的签名（需要运用 UTF-8 字符集）。

PYTHON 连接钉钉传输工作数据监控

https://oapi.dingtalk.com/robot/send?access_token=XXXXXX&timestamp=XXX&sign=XXX

PYTHON 连接钉钉传输工作数据监控

PYTHON 连接钉钉传输工作数据监控

PYTHON 连接钉钉传输工作数据监控

测验机器人

python 代码版别

这儿直接运用 加签版别的,因为这种时刻断定的才是最常用的

import requests
import json
import time
import hmac
import hashlib
import base64
import urllib.parse
timestamp = str(round(time.time() * 1000))   #毫秒级时刻戳
secret = '你的secret'
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))
#往post接口里面仍数据
def dingmessage():
# 恳求的URL，WebHook地址
    webhook = "https://oapi.dingtalk.com/robot/send?access_token=你的token&timestamp="+str(timestamp)+"&sign="+str(sign)
    print(webhook)
#构建恳求头部
    header = {
        "Content-Type": "application/json",
        "Charset": "UTF-8"
}
#构建恳求数据
    tex = "能力越大，职责越大。"
    message ={
        "msgtype": "text",
        "text": {
            "content": tex
        },
        "at": {
            "isAtAll": False
        }
    }
#对恳求的数据进行json封装
    message_json = json.dumps(message)
#发送恳求
    info = requests.post(url=webhook,data=message_json,headers=header)
#打印返回的成果
    print(info.text)
if __name__=="__main__":
    dingmessage()

打印成果

PYTHON 连接钉钉传输工作数据监控

假如显示 ok 那么便是说数据传输成功了

这时候接入钉钉即可 检查数据

PYTHON 连接钉钉传输工作数据监控

好了现在 py 现已可以给钉钉发送信息了,那么怎么做监控呢

创建时刻监控 实时发送信息

我这儿以 apscheduler 框架 进行定时巡回

一、安装 APScheduler

pip install apscheduler

复制代码

二、基本概念

APScheduler 有四大组件：

1、触发器 triggers ：触发器包括调度逻辑。每个作业都有自己的触发器，用于确定下一个使命何时运转。除了初始配置之外，触发器是完全无状态的。有三种内建的 trigger:

（1）date: 特定的时刻点触发

（2）interval: 固定时刻间隔触发

(3）cron: 在特定时刻周期性地触发

2、使命储存器 job stores：用于寄存使命，把使命寄存在内存（为默许 MemoryJobStore）或数据库中。3、执行器 executors： 执行器是将使命提交到线程池或进程池中运转，当使命完结时，执行器告诉调度器触发相应的事情。

4、调度器 schedulers： 把上方三个组件作为参数，经过创建调度器实例来运转根据开发需求挑选相应的组件，下面是不同的调度器组件：BlockingScheduler 堵塞式调度器：适用于只跑调度器的程序。BackgroundScheduler 后台调度器：适用于非堵塞的状况，调度器会在后台独立运转。AsyncIOScheduler AsyncIO 调度器，适用于运用运用 AsnycIO 的状况。GeventScheduler Gevent 调度器，适用于运用经过 Gevent 的状况。TornadoScheduler Tornado 调度器，适用于构建 Tornado 运用。TwistedScheduler Twisted 调度器，适用于构建 Twisted 运用。QtScheduler Qt 调度器，适用于构建 Qt 运用。

三、运用步骤

1、新建一个调度器 schedulers

2、增加调度使命

3、运转调度使命

PYTHON 连接钉钉传输工作数据监控

PYTHON 连接钉钉传输工作数据监控

PYTHON 连接钉钉传输工作数据监控

比方写 3 便是说 冯 3 触发

*/3 便是说 每隔 3 单位触发

假设触发条件

咱们这儿断定假如当前时刻 4 个数据加起来大于 12 那么 就发送音讯

每分钟发送一次

from apscheduler.schedulers.blocking import BlockingScheduler
import datetime
# 输出时刻
import pymssql
import copy
import time
import requests
import json
import hmac
import hashlib
import base64
import urllib.parse
timestamp = str(round(time.time() * 1000))
secret = '你的secret'
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))
# now = time.time() #返回float数据
#
#
# #毫秒级时刻戳
# print(int(round(now * 1000)))
# now1=int(round(now * 1000))
def dingmessage(num):
# 恳求的URL，WebHook地址
    webhook = "https://oapi.dingtalk.com/robot/send?access_token=你的token&timestamp="+str(timestamp)+"&sign="+str(sign)
    print(webhook)
#构建恳求头部
    header = {
        "Content-Type": "application/json",
        "Charset": "UTF-8"
}
#构建恳求数据
    tex = "能力越大，职责越大。"+str(num)
    message ={
        "msgtype": "text",
        "text": {
            "content": tex
        },
        "at": {
            "isAtAll": False
        }
    }
#对恳求的数据进行json封装
    message_json = json.dumps(message)
#发送恳求
    info = requests.post(url=webhook,data=message_json,headers=header)
#打印返回的成果
    print(info.text)
def job():
    hour = datetime.datetime.now().hour
    min = datetime.datetime.now().minute
    f1 = str(hour)[0]
    f2 = str(hour)[1]
    f3 = str(min)[0]
    f4 = str(min)[1]
    print(f1)
    print(f2)
    print(f3)
    print(f4)
    print(int(f1) + int(f2) + int(f3) + int(f4))
    if int(f1) + int(f2) + int(f3) + int(f4) > 12:
        dingmessage(int(f1) + int(f2) + int(f3) + int(f4) )
# BlockingScheduler
scheduler = BlockingScheduler()
#1-23小时内,每一分钟触发一次, 后边那个意思是说假如3600 秒触发 那么讲从头执行该使命
scheduler.add_job(job, 'cron', hour='1-23', minute='*/1',misfire_grace_time=3600)#1
scheduler.start()

成果

运转成果

PYTHON 连接钉钉传输工作数据监控

钉钉成果

如图 23.18 分 2+3+1+8=14 >12 触发条件 就给钉钉 发送信息了

PYTHON 连接钉钉传输工作数据监控

致此一个见得 发送文字功用实现,有问题即可发送钉钉

还有一点便是钉钉一定要调成前台答应显示, 然后在录音个对应的提示. 懒人必备,